Agent loop¶
@channel keeps a duplex session open; @handle is a tool on the fabric.
The agent loop is the glue: the model plans, tools run as query_once on key
expressions, results go back into the conversation, and the model answers.
An agent is a service on a key — not an in-process graph. Tools are other services on the same mesh (or local callables for tests).
Pieces¶
| Piece | Role |
|---|---|
MeshTool / tools_from_handlers |
Catalogue of callable mesh endpoints |
tools_from_discovery |
The same catalogue, read off the fabric's manifests |
Model / OpenAIChatModel |
One completion turn (OpenAI-compatible /v1/chat/completions) |
run_agent |
plan → tool → observe until text or max_steps |
drive_channel |
Reload durable history, then run the loop per inbound turn |
app.approvals() |
Human approval before an irreversible tool runs |
from istos import Istos, ChannelSession
from istos.agent import OpenAIChatModel, drive_channel, tools_from_handlers
app = Istos(http_port=8080)
@app.handle("math/add")
async def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b
tools = tools_from_handlers(app, prefixes=["math/add"])
model = OpenAIChatModel(
base_url="http://127.0.0.1:1234/v1",
model="qwen/qwen3.5-9b",
)
@app.channel("agent/chat", ws="/chat", durable=True)
async def chat(s: ChannelSession):
await drive_channel(
s, model, tools,
system="You are a calculator assistant. Use math/add when needed.",
)
if __name__ == "__main__":
app.run()
tools_from_handlers builds the same name / docstring / JSON Schema catalogue
MCP uses (math/add → tool math-add). Pass prefixes= to whitelist. Plumbing
under .istos/ is skipped.
Events¶
run_agent yields AgentEvent values. drive_channel sends each (except
done) as a JSON object on the session:
kind |
Meaning |
|---|---|
tool_call |
Model asked to run a tool (name, arguments, tool_call_id) |
approval_request |
Waiting on a human before a gated tool runs (approval_id) |
approval_decision |
The human answered; error=True when refused |
tool_result |
Tool returned (content); error=True when it raised |
message |
Final assistant text for this turn (content) |
done |
Turn finished (not sent on the channel) |
async for event in run_agent(model, tools, messages, token=jwt):
if event.kind == "message":
print(event.content)
messages is mutated in place so you can keep a multi-turn list across
run_agent calls. Mesh tool calls forward token= on query_once, so the
tool's authorizer still runs.
drive_channel(..., send_events=False) sends only the final message content
(plain string) — useful when the client does not want tool-call frames.
Remote tools¶
A tool does not have to live on the same process. Point MeshTool at another
node's prefix; query_once finds it on the fabric:
from istos.agent import MeshTool
tools = [
MeshTool(
"billing/invoice",
app=app,
description="Create an invoice",
parameters={
"type": "object",
"properties": {
"customer": {"type": "string"},
"amount": {"type": "number"},
},
"required": ["customer", "amount"],
},
),
]
Writing another service's schema by hand goes stale. That schema is already published — capability discovery serves it — so read it instead:
from istos import tools_from_discovery
# Every remote @handle, with the owner's schema and docstring. Needs an open
# session, so call it from a lifespan or a handler.
tools = await tools_from_discovery(app, services=["billing", "search"])
Only handle entries become tools — a mesh tool is a query_once, so streams
and channels are skipped. prefixes= whitelists exact keys. The result is a
snapshot: call it again to pick up nodes that joined later.
Human approval for irreversible tools¶
A tool that moves money or deletes data should not fire because a model felt like it. The endpoint's owner declares the requirement, so every agent that discovers the tool inherits it:
@app.handle("billing/refund", approval="moves real money")
async def refund(order_id: str) -> dict:
"""Refund an order."""
...
The agent node passes a gate to the loop:
gate = app.approvals(timeout_s=600, authorizer=require_roles("ops"))
@app.channel("agent/chat", ws="/chat", durable=True)
async def chat(s: ChannelSession):
await drive_channel(s, model, tools, approvals=gate)
Now a call to billing/refund suspends the turn: an approval_request event
goes out (with approval_id), and the tool runs only after a yes. Nothing polls
— the waiting agent holds an asyncio.Event that the decision wakes.
The decision arrives over the fabric, from any node:
from istos import decide_approval, list_approvals
for req in await list_approvals(app):
print(req["id"], req["tool"], req["arguments"], req["reason"])
await decide_approval(app, req["id"], approved=True, by="amir")
# or correct it instead of refusing outright:
await gate.approve(req["id"], by="amir", arguments={"order_id": "o-42"})
list_approvals asks .istos/approvals/* and decide_approval asks
.istos/approvals/*/decide; each waiting node has keys of its own, so the
wildcard reaches all of them and only the node holding the request acts. Put the
HTTP gateway in front for a browser UI.
Fail-closed, deliberately:
- A denial or a timeout never runs the tool. Both come back to the model as a
failed
tool_resultcarrying the reason, so it can tell the user rather than retry blindly. - An undecided request expires after
timeout_s(Nonewaits forever). - Passing a gated tool with no gate raises
ValueErrorat the start of the run — a missing gate can never read as a pass. approval=is advisory: it tells agents to stop, it does not stop a peer that queries the key directly. Keep the handler'sauthorizeras the real gate. Deciding is privileged too, so giveapp.approvals()an authorizer — Istos warns withIstosSecurityWarningwhen the decide key is left open.
A caller can also gate a prefix its owner did not declare:
tools_from_discovery(app, approval=["search/purge"]).
Requests are written through to the app's storage, so with Redis or SQLAlchemy an operator can still list what was outstanding after a restart. The waiter itself is in-memory by nature: an agent that crashed is no longer waiting, so its recovered request is stale and expires.
Own model¶
Anything with async def complete(messages, *, tools=None) -> ModelReply works.
OpenAIChatModel is the battery for OpenAI, LM Studio, vLLM, and similar. Tool
call arguments are parsed from the usual OpenAI tool_calls shape.
Handoff between agents¶
For a triage-and-specialists setup, give each role its own Agent — a model,
its tools, a system prompt, and the agents it may hand off to — and drive the
channel with drive_agents. The model transfers by calling a synthetic
transfer_to_<name> tool; the loop swaps the active agent but keeps the shared
message history, so context carries across.
from istos.agent import Agent, drive_agents
billing = Agent(
"billing", model, tools=[refund_tool],
system="You handle refunds.", description="refunds and billing questions",
)
router = Agent(
"router", model, handoffs=[billing],
system="Route the user to the right specialist.",
)
billing.handoffs = [router] # return handoff: hand back when done
@app.channel("agent/chat", durable=True)
async def chat(s):
await drive_agents(s, router, token=jwt) # forwarded to every tool call
- Handoff graphs may cycle, so a specialist can hand back to the router (triage → specialist → triage).
- The active agent persists across turns within a session, and is restored on
reconnect from persisted
handoffframes (send_events=True); otherwise a resumed session restarts at the entry agent. tokenforwards to whichever agent's tools run, so authorizers see the original principal no matter how many handoffs occurred.- A remote specialist on another node is reached as a mesh tool
(
query_onceon its key), not an in-process handoff — handoff switches the local driving agent; the mesh is how you call across nodes.
Tracing¶
With Istos(enable_tracing=True) the loop emits an istos.agent.completion span
per model turn and an istos.agent.tool span per tool call, nested under the
channel handler's request span — so an agent's model and tool work shows up
inside the same distributed trace that already spans hops over Zenoh.
Completion spans carry GenAI attributes (gen_ai.response.model,
gen_ai.usage.input_tokens / output_tokens, finish reason) and, for
multi-agent, the active istos.agent.name. Token counts come from ModelReply
(model / finish_reason / usage), which OpenAIChatModel fills from the
response; a custom model that leaves them unset simply omits those attributes.
Tool spans carry the tool name and prefix and flag errors. Everything is a no-op
until tracing is configured, so OpenTelemetry stays an optional dependency.
Honest limits¶
- The loop is not a DAG engine. Branching and long-running workflows stay on
queues (
chain/group/chord) or your own control flow. - Durable channel history reconstructs the full tool transcript on reconnect
(assistant
tool_calls+toolresults), so a resumed session keeps its tool context. A tool call with no recorded result (a crash mid-tool) is dropped to keep the message sequence valid; passinclude_tools=Falsefor text only. OpenAIChatModelis non-streaming completions. For token streaming without tools, keep using@stream/stream_queryas before.- MCP and
tools_from_handlersshare the catalogue idea; MCP still lists this node's@handleonly. An agent can call remote prefixes that MCP on this node does not advertise. - An approval waits in the process that filed it. Run several replicas of one agent service and each holds its own pending set — which the fan-out decide handles (every replica is asked, the holder answers), but a replica that dies takes its waiter with it.
See also: Channels, MCP, agent channel recipe, agent with tools recipe.