Agent API¶

Mesh tool catalogue, model protocol, and the plan → tool → observe loop.

Agent loop over mesh tools — plan → query_once → observe.

The fabric primitives stay as they are (@channel, @handle, queues). This package is the glue that turns a channel handler into an agent whose tools are other services on Zenoh.

`Agent` `dataclass` ¶

One agent in a handoff graph: a model, its tools, a system prompt, and the agents it may transfer to.

name must be usable in a tool name ([A-Za-z0-9_-]); it is sanitized the same way key expressions are. description is surfaced to a router model in the transfer_to_<name> tool so it knows when to hand off.

Source code in src/istos/agent/multi.py

@dataclass
class Agent:
    """One agent in a handoff graph: a model, its tools, a system prompt, and the
    agents it may transfer to.

    ``name`` must be usable in a tool name (``[A-Za-z0-9_-]``); it is sanitized
    the same way key expressions are. ``description`` is surfaced to a router
    model in the ``transfer_to_<name>`` tool so it knows when to hand off.
    """

    name: str
    model: Model
    tools: Sequence[MeshTool] = field(default_factory=list)
    system: Optional[str] = None
    handoffs: Sequence["Agent"] = field(default_factory=list)
    description: str = ""

    @property
    def tool_name(self) -> str:
        return tool_name(self.name)

`AgentEvent` `dataclass` ¶

One step the loop emits for the caller to forward (e.g. over a channel).

kind is one of:

message — final assistant text for this turn
tool_call — model asked to run a mesh tool (name, arguments)
tool_result — tool returned (content); error when it raised
handoff — active agent transferred to name (multi-agent loop)
done — turn finished (no more steps)

Source code in src/istos/agent/loop.py

@dataclass(frozen=True)
class AgentEvent:
    """One step the loop emits for the caller to forward (e.g. over a channel).

    ``kind`` is one of:

    - ``message`` — final assistant text for this turn
    - ``tool_call`` — model asked to run a mesh tool (``name``, ``arguments``)
    - ``tool_result`` — tool returned (``content``); ``error`` when it raised
    - ``handoff`` — active agent transferred to ``name`` (multi-agent loop)
    - ``done`` — turn finished (no more steps)
    """

    kind: str
    content: Any = None
    name: Optional[str] = None
    arguments: Optional[Dict[str, Any]] = None
    tool_call_id: Optional[str] = None
    error: bool = False

`MeshTool` ¶

One mesh endpoint the agent may call.

Built from a local @handle via :func:tools_from_handlers, or by hand when the tool lives on another node (pass app and the remote prefix)::

MeshTool("math/add", app=app, description="Add two integers",
         parameters={"type": "object", "properties": {
             "a": {"type": "integer"}, "b": {"type": "integer"},
         }, "required": ["a", "b"]})

Source code in src/istos/agent/tools.py

class MeshTool:
    """One mesh endpoint the agent may call.

    Built from a local ``@handle`` via :func:`tools_from_handlers`, or by hand
    when the tool lives on another node (pass ``app`` and the remote prefix)::

        MeshTool("math/add", app=app, description="Add two integers",
                 parameters={"type": "object", "properties": {
                     "a": {"type": "integer"}, "b": {"type": "integer"},
                 }, "required": ["a", "b"]})
    """

    def __init__(
        self,
        prefix: str,
        *,
        app: Any = None,
        name: Optional[str] = None,
        description: str = "",
        parameters: Optional[dict] = None,
        invoke: Optional[Callable[..., Awaitable[Any]]] = None,
    ) -> None:
        if app is None and invoke is None:
            raise ValueError("MeshTool needs an app (for query_once) or an invoke callable")
        self.prefix = prefix
        self.name = name or tool_name(prefix)
        self.description = description or self.name
        self.parameters = parameters or {"type": "object", "properties": {}}
        self._app = app
        self._invoke = invoke

    def openai_schema(self) -> dict:
        """Tool definition in the OpenAI chat-completions shape."""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

    async def call(
        self,
        arguments: dict,
        *,
        token: Optional[Union[bytes, str]] = None,
        timeout_s: float = 5.0,
    ) -> Any:
        """Run the tool. Mesh tools go through ``query_once`` so authorizers run."""
        if self._invoke is not None:
            return await self._invoke(**arguments)
        assert self._app is not None
        return await self._app.query_once(
            self.prefix, token=token, timeout_s=timeout_s, **arguments
        )

`call(arguments, *, token=None, timeout_s=5.0)` `async` ¶

Run the tool. Mesh tools go through query_once so authorizers run.

Source code in src/istos/agent/tools.py

async def call(
    self,
    arguments: dict,
    *,
    token: Optional[Union[bytes, str]] = None,
    timeout_s: float = 5.0,
) -> Any:
    """Run the tool. Mesh tools go through ``query_once`` so authorizers run."""
    if self._invoke is not None:
        return await self._invoke(**arguments)
    assert self._app is not None
    return await self._app.query_once(
        self.prefix, token=token, timeout_s=timeout_s, **arguments
    )

`openai_schema()` ¶

Tool definition in the OpenAI chat-completions shape.

Source code in src/istos/agent/tools.py

def openai_schema(self) -> dict:
    """Tool definition in the OpenAI chat-completions shape."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }