feat: add AgentOSClient class and cookbooks #5613
Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds a comprehensive HTTP client (AgentOSClient) for interacting with remote AgentOS instances, along with extensive unit tests and cookbook examples demonstrating its usage.
Key Changes:
- New
AgentOSClientclass providing async HTTP client for all AgentOS API operations - Comprehensive unit tests with 100+ test cases covering initialization, HTTP methods, and all API operations
- Five cookbook examples demonstrating client usage patterns (basic connection, agent runs, memory operations, session management, knowledge search)
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
libs/agno/agno/os/client.py |
Core client implementation with 2259 lines providing async methods for all AgentOS operations (discovery, runs, sessions, memory, knowledge, evals) |
libs/agno/agno/os/__init__.py |
Exports AgentOSClient alongside existing AgentOS class |
libs/agno/tests/unit/os/test_client.py |
Comprehensive unit tests (666 lines) covering all client functionality with mocked HTTP calls |
libs/agno/tests/unit/os/__init__.py |
Package initialization for unit tests |
cookbook/agent_os/client/*.py |
Five working examples demonstrating client usage patterns |
cookbook/agent_os/client/README.md |
Detailed documentation with API reference and usage examples |
cookbook/agent_os/client/__init__.py |
Package initialization for cookbook examples |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
libs/agno/agno/os/client.py
Outdated
| Raises: | ||
| HTTPStatusError: On HTTP errors | ||
| """ | ||
| import json |
There was a problem hiding this comment.
[nitpick] The json module is imported inside 10 different methods. Consider moving this import to the top of the file to improve code organization and avoid repeated import overhead. While Python caches module imports, having it at the module level is clearer and follows common Python conventions.
libs/agno/agno/os/client.py
Outdated
| if table: | ||
| params["table"] = table | ||
|
|
||
| endpoint = "/memories?" + "&".join(f"{k}={v}" for k, v in params.items()) |
There was a problem hiding this comment.
Query parameters are concatenated directly without URL encoding. This could cause issues if parameter values contain special characters (e.g., spaces, ampersands, equals signs). Consider using urllib.parse.urlencode() or httpx's built-in parameter support instead of manual string concatenation. For example:
# Instead of:
endpoint = "/memories?" + "&".join(f"{k}={v}" for k, v in params.items())
# Use:
from urllib.parse import urlencode
endpoint = f"/memories?{urlencode(params)}"
# Or let httpx handle it:
response = await self._http_client.get(url, params=params, headers=headers)
libs/agno/agno/os/client.py
Outdated
| VectorSearchResult, | ||
| ) | ||
| from agno.os.routers.memory.schemas import ( | ||
| UserMemoryCreateSchema, |
There was a problem hiding this comment.
Import of 'UserMemoryCreateSchema' is not used.
| UserMemoryCreateSchema, |
| from agno.os.schema import ( | ||
| AgentResponse, | ||
| AgentSessionDetailSchema, | ||
| AgentSummaryResponse, | ||
| ConfigResponse, | ||
| CreateSessionRequest, | ||
| DeleteSessionRequest, | ||
| Model, | ||
| PaginatedResponse, | ||
| RunSchema, | ||
| SessionSchema, | ||
| TeamResponse, | ||
| TeamRunSchema, | ||
| TeamSessionDetailSchema, | ||
| TeamSummaryResponse, | ||
| UpdateSessionRequest, | ||
| WorkflowResponse, | ||
| WorkflowRunSchema, | ||
| WorkflowSessionDetailSchema, | ||
| WorkflowSummaryResponse, | ||
| ) |
There was a problem hiding this comment.
Import of 'CreateSessionRequest' is not used.
Import of 'DeleteSessionRequest' is not used.
Import of 'UpdateSessionRequest' is not used.
libs/agno/agno/os/client.py
Outdated
| WorkflowSummaryResponse, | ||
| ) | ||
| try: | ||
| from httpx import AsyncClient, HTTPStatusError |
There was a problem hiding this comment.
Import of 'HTTPStatusError' is not used.
| from httpx import AsyncClient, HTTPStatusError | |
| from httpx import AsyncClient |
…tent uploads with AgentOSClient.
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>
libs/agno/agno/os/client.py
Outdated
| self, | ||
| base_url: str, | ||
| api_key: Optional[str] = None, | ||
| timeout: float = 300.0, |
There was a problem hiding this comment.
300 seconds? that's a lot probably. Also okay to have int here, we use that everywhere for seconds I think, cool to keep consistency
libs/agno/agno/os/client.py
Outdated
| """Exit async context manager and cleanup resources.""" | ||
| await self.close() | ||
|
|
||
| async def connect(self) -> "AgentOSClient": |
There was a problem hiding this comment.
We need the instance to call this, so no need to return the instance itself
| self._http_client = AsyncClient(timeout=self.timeout) | ||
| return self | ||
|
|
||
| async def close(self) -> None: |
There was a problem hiding this comment.
I think OK to not carry sync methods here, but not sure anymore... @dirkbrnd what do you think?
There was a problem hiding this comment.
We'll just need sync/async for the connection/client pieces. I can explain later, but we'll need sync client for some of the methods.
libs/agno/agno/os/client.py
Outdated
| async for line in response.aiter_lines(): | ||
| yield line | ||
|
|
||
| async def _post_empty(self, endpoint: str) -> bool: |
There was a problem hiding this comment.
These methods look a bit too similar, we can probably simplify and just have get, post, delete?
| data = await self._get("/agents") | ||
| return [AgentSummaryResponse.model_validate(item) for item in data] | ||
|
|
||
| async def get_agent(self, agent_id: str) -> AgentResponse: |
There was a problem hiding this comment.
- let's keep the run_agent and run_team methods together with these ones?
- missing workflow endpoints i think
There was a problem hiding this comment.
which workflow endpoint am i missing over here? can you please specify?
libs/agno/agno/os/client.py
Outdated
| if agent_id: | ||
| payload["agent_id"] = agent_id | ||
| if team_id: | ||
| payload["team_id"] = team_id | ||
| if model_id: | ||
| payload["model_id"] = model_id | ||
| if model_provider: | ||
| payload["model_provider"] = model_provider | ||
| if expected_output: | ||
| payload["expected_output"] = expected_output | ||
| if expected_tool_calls: | ||
| payload["expected_tool_calls"] = expected_tool_calls | ||
| if num_iterations != 1: | ||
| payload["num_iterations"] = num_iterations |
There was a problem hiding this comment.
payload can take Nones, right? I think no need to do this
If we do it, we will basically force ourselves to maintain this separately from the schema class itself. That's a bit of a smell and will make updates here a bit more risky (we will at some point forget about one of the places to update)
| a remote AgentOS instance and perform basic operations. | ||
|
|
||
| Prerequisites: | ||
| 1. Start an AgentOS server: |
There was a problem hiding this comment.
Maybe make a helper file in this directory to spin up an AgentOS that would work for all the cookbooks
| self.base_url = base_url.rstrip("/") | ||
| self.api_key = api_key or getenv("AGNO_API_KEY") | ||
| self.timeout = timeout | ||
| self._http_client: Optional[AsyncClient] = None |
There was a problem hiding this comment.
We'll need a sync and async clients. We'll use them selectively, so no need to duplicate all the methods, but we'll need both, and then also enter, etc for setting up the connection
| Raises: | ||
| HTTPStatusError: On HTTP errors (4xx, 5xx) | ||
| """ | ||
| if not self._http_client: |
There was a problem hiding this comment.
Rather make a _get_client() that returns or creates the client. You can use it above in the connect method as well
libs/agno/agno/os/client.py
Outdated
|
|
||
| async def __aenter__(self) -> "AgentOSClient": | ||
| """Enter async context manager.""" | ||
| self._http_client = AsyncClient(timeout=self.timeout) |
There was a problem hiding this comment.
Shouldn't you call connect here?
libs/agno/agno/os/client.py
Outdated
| data["context"] = json.dumps(context) | ||
|
|
||
| response_data = await self._post_form_data(endpoint, data) | ||
| return RunSchema.model_validate(response_data) |
There was a problem hiding this comment.
RunSchema is the API response schema. and i see that RunOutput is the internal agent run output class. Now since AgentOSclient is just an API client
that receives JSON responses and parses them, so imo
RunSchema is correct for deserializing the API response.
Please correct me if im wrong and we should use RunOutput here
There was a problem hiding this comment.
/agents/{agent_id}/runs does not return RunSchema... RunSchema is used by the sessions endpoint when returning a list of stored runs.
The runs endpoint returns either a run output object or a stream of run events. And this client should do the same
Summary
Added the AgentOsClient class which is just a http client can be used to connect to remote agno agents.
(If applicable, issue number: #____)
Type of change
Checklist
./scripts/format.shand./scripts/validate.sh)Additional Notes
Add any important context (deployment instructions, screenshots, security considerations, etc.)