"""Agents resource for Taimoe Platform SDK."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from uuid import UUID
from taimoe.platform.types.agent import Agent, AgentBinding, AgentCreatePayload
from taimoe.platform.types.sync import AgentRuntimeConfig
if TYPE_CHECKING:
from taimoe.platform.client.client import TaimoeClient
def _validate_kb_slug(slug: str) -> None:
"""Raise ValueError if `slug` is not in 'team/kb' form."""
if "/" not in slug:
raise ValueError(f"Invalid kb_slug '{slug}'. Expected format 'team/kb_slug'.")
team, kb = slug.split("/", 1)
if not team or not kb:
raise ValueError(f"Invalid kb_slug '{slug}'. Expected format 'team/kb_slug'.")
[docs]
class AgentsResource:
"""Resource for interacting with /api/v1/agents endpoints."""
def __init__(self, client: TaimoeClient) -> None:
self._client = client
[docs]
def get_runtime_config_by_name(self, name: str) -> AgentRuntimeConfig:
"""Fetch a single agent's runtime config by `runtime_agent_id`
(or `name`). Runtime-less: no Runtime binding required."""
response = self._client._request("GET", f"/api/v1/agents/by-name/{name}")
return AgentRuntimeConfig.model_validate(response)
[docs]
async def get_runtime_config_by_name_async(self, name: str) -> AgentRuntimeConfig:
response = await self._client._request_async("GET", f"/api/v1/agents/by-name/{name}")
return AgentRuntimeConfig.model_validate(response)
[docs]
def create(
self,
payload: AgentCreatePayload,
*,
knowledge_bases: list[str] | None = None,
org: str | None = None,
team: str = "default",
) -> Agent:
"""Create a new agent and optionally bind knowledge bases to it.
Backend `POST /organizations/{org}/teams/{team}/agents` does not accept
knowledge bases in the create body, so when `knowledge_bases` is
provided the SDK issues one bind call per slug after creation.
Args:
payload: Typed agent create payload.
knowledge_bases: Optional list of 'team/kb' slugs to bind.
org: Organization slug. Falls back to the client's org.
team: Team slug.
"""
org_name = org or self._client.org
path = f"/api/v1/organizations/{org_name}/teams/{team}/agents"
body = payload.model_dump(exclude_none=True)
agent = Agent.model_validate(self._client._request("POST", path, json=body))
for slug in knowledge_bases or ():
self.bind_knowledge_base(agent.id, slug)
return agent
[docs]
async def create_async(
self,
payload: AgentCreatePayload,
*,
knowledge_bases: list[str] | None = None,
org: str | None = None,
team: str = "default",
) -> Agent:
"""Create a new agent asynchronously."""
org_name = org or self._client.org
path = f"/api/v1/organizations/{org_name}/teams/{team}/agents"
body = payload.model_dump(exclude_none=True)
response = await self._client._request_async("POST", path, json=body)
agent = Agent.model_validate(response)
for slug in knowledge_bases or ():
await self.bind_knowledge_base_async(agent.id, slug)
return agent
[docs]
def bind_knowledge_base(self, agent_id: str, kb_slug: str) -> AgentBinding:
"""Bind a knowledge base to an agent using its 'team/kb' slug.
The backend resolves the slug via `kb_resolver`, so no org/team scoping
is needed in the URL.
"""
_validate_kb_slug(kb_slug)
response = self._client._request(
"POST",
f"/api/v1/agents/{agent_id}/knowledge-bases",
json={"slug": kb_slug},
)
return AgentBinding.model_validate(response)
[docs]
async def bind_knowledge_base_async(self, agent_id: str, kb_slug: str) -> AgentBinding:
"""Bind a knowledge base to an agent asynchronously."""
_validate_kb_slug(kb_slug)
response = await self._client._request_async(
"POST",
f"/api/v1/agents/{agent_id}/knowledge-bases",
json={"slug": kb_slug},
)
return AgentBinding.model_validate(response)
[docs]
def unbind_knowledge_base(self, agent_id: str, kb_id: str | UUID) -> None:
"""Unbind a knowledge base from an agent by its kb UUID.
Backend responds with 204 No Content.
"""
self._client._request("DELETE", f"/api/v1/agents/{agent_id}/knowledge-bases/{kb_id}")
[docs]
async def unbind_knowledge_base_async(self, agent_id: str, kb_id: str | UUID) -> None:
"""Unbind a knowledge base from an agent asynchronously."""
await self._client._request_async(
"DELETE",
f"/api/v1/agents/{agent_id}/knowledge-bases/{kb_id}",
)