Skip to content

Agent API

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

Details: Agent 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 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@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
  • approval_request — waiting on a human before a gated tool runs; approval_id is what to decide, content the reason if any
  • approval_decision — the human answered; error when refused
  • handoff — active agent transferred to name (multi-agent loop)
  • done — turn finished (no more steps)
Source code in src/istos/agent/loop.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@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
    - ``approval_request`` — waiting on a human before a gated tool runs;
      ``approval_id`` is what to decide, ``content`` the reason if any
    - ``approval_decision`` — the human answered; ``error`` when refused
    - ``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
    approval_id: Optional[str] = None

ApprovalDenied

Bases: IstosError

A human refused the tool call.

Source code in src/istos/agent/approval.py 
53
54
55
56
57
class ApprovalDenied(IstosError):
    """A human refused the tool call."""

    def __init__(self, message: str = "Tool call denied", **kwargs: Any):
        super().__init__(message, code="approval_denied", status=403, **kwargs)

ApprovalGate

Pending approvals for one node: durable state plus the waiters.

Built by :meth:Istos.approvals, which also registers the two fabric keys. Construct one directly only when driving it yourself (a test, or an agent that is not an Istos app).

State is written through to the app's StoragePlugin, so with Redis or SQLAlchemy a restart can still list and settle what was outstanding; with the in-memory default the pending set dies with the process. The waiter itself is always in-memory — an agent that has crashed is no longer waiting, and its recovered request is stale by definition.

Source code in src/istos/agent/approval.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
class ApprovalGate:
    """Pending approvals for one node: durable state plus the waiters.

    Built by :meth:`Istos.approvals`, which also registers the two fabric keys.
    Construct one directly only when driving it yourself (a test, or an agent
    that is not an Istos app).

    State is written through to the app's ``StoragePlugin``, so with Redis or
    SQLAlchemy a restart can still list and settle what was outstanding; with the
    in-memory default the pending set dies with the process. The waiter itself is
    always in-memory — an agent that has crashed is no longer waiting, and its
    recovered request is stale by definition.
    """

    def __init__(
        self,
        *,
        storage: Any = None,
        timeout_s: Optional[float] = 300.0,
        service_name: str = "istos",
        node_id: Optional[str] = None,
        on_request: Optional[Callable[[ApprovalRequest], Union[None, Awaitable[None]]]] = None,
    ) -> None:
        self._storage = storage
        self.timeout_s = timeout_s
        self._service = service_name
        # A chunk of its own per process: replicas of one service would otherwise
        # share a key, and `@handle` answers a shared key from exactly one of them.
        self._node = node_id or uuid.uuid4().hex[:8]
        # Called with each new request — notify a dashboard, page someone. Public
        # so it can be set after construction (a notifier often needs the gate).
        self.on_request = on_request
        self._requests: Dict[str, ApprovalRequest] = {}
        self._waiters: Dict[str, asyncio.Event] = {}
        self._lock = asyncio.Lock()
        self._loaded = False

    # --- fabric keys ---

    @property
    def key(self) -> str:
        """Where this node lists its pending requests."""
        return f"{APPROVALS_KEY}/{key_chunk(self._service)}-{self._node}"

    @property
    def decide_key(self) -> str:
        """Where this node accepts decisions."""
        return f"{self.key}/decide"

    # --- persistence ---

    def _index_key(self) -> str:
        return f"approvals:{key_chunk(self._service)}:index"

    def _req_key(self, request_id: str) -> str:
        return f"approvals:{key_chunk(self._service)}:req:{request_id}"

    async def _write(self, req: ApprovalRequest) -> None:
        if self._storage is None:
            return
        try:
            await self._storage.put(self._req_key(req.id), req.to_dict())
            pending = sorted(
                r.id for r in self._requests.values() if r.state == ApprovalState.PENDING
            )
            await self._storage.put(self._index_key(), pending)
        except Exception:
            _logger.exception("Could not persist approval %s", req.id)

    async def load(self) -> None:
        """Recover requests still pending from a previous run. Runs once."""
        if self._loaded:
            return
        self._loaded = True
        if self._storage is None:
            return
        try:
            for request_id in await self._storage.get(self._index_key()) or []:
                raw = await self._storage.get(self._req_key(request_id))
                if isinstance(raw, dict) and request_id not in self._requests:
                    self._requests[request_id] = ApprovalRequest.from_dict(raw)
        except Exception:  # recovery is best-effort — never block the gate
            _logger.exception("Could not recover pending approvals")

    # --- filing and settling ---

    async def request(
        self,
        tool: str,
        arguments: Optional[Dict[str, Any]] = None,
        *,
        prefix: Optional[str] = None,
        reason: Optional[str] = None,
        requester: Optional[str] = None,
        conversation_id: Optional[str] = None,
        timeout_s: Optional[float] = -1.0,
    ) -> ApprovalRequest:
        """File a request and return it, without waiting. ``timeout_s`` defaults
        to the gate's; pass ``None`` for no deadline."""
        await self.load()
        ttl = self.timeout_s if timeout_s == -1.0 else timeout_s
        req = ApprovalRequest(
            id=uuid.uuid4().hex,
            tool=tool,
            arguments=dict(arguments or {}),
            prefix=prefix,
            reason=reason,
            requester=requester or self._service,
            conversation_id=conversation_id,
            expires_at=time.time() + ttl if ttl else 0.0,
        )
        async with self._lock:
            self._requests[req.id] = req
            self._waiters[req.id] = asyncio.Event()
        await self._write(req)
        _logger.info(
            "Approval %s pending for tool %s", req.id, tool,
            extra={"approval_id": req.id, "tool": tool, "conversation_id": conversation_id},
        )
        if self.on_request is not None:
            try:
                result = self.on_request(req)
                if asyncio.iscoroutine(result):
                    await result
            except Exception:  # a broken notifier must not strand the request
                _logger.exception("approval on_request callback failed")
        return req

    async def wait(
        self, request_id: str, *, timeout_s: Optional[float] = -1.0,
    ) -> ApprovalRequest:
        """Block until the request is settled.

        Returns the approved request (its ``arguments`` are what to run — an
        approver may have edited them). Raises :class:`ApprovalDenied` on a no,
        :class:`ApprovalTimeout` when the deadline passes, and
        :class:`NotFoundError` for an unknown id.
        """
        async with self._lock:
            req = self._requests.get(request_id)
            event = self._waiters.get(request_id)
        if req is None:
            raise NotFoundError(f"No approval request {request_id!r}")
        if req.state != ApprovalState.PENDING:
            return self._settled(req)

        # The gate's own deadline and the caller's, whichever comes first.
        budget = self.timeout_s if timeout_s == -1.0 else timeout_s
        if req.expires_at:
            remaining = req.expires_at - time.time()
            budget = remaining if budget is None else min(budget, remaining)
        if event is None:  # recovered from storage: nobody in this process waits
            raise ApprovalTimeout(
                f"Approval {request_id} was filed by an earlier run of this node",
                details={"id": request_id},
            )
        try:
            if budget is not None and budget <= 0:
                raise asyncio.TimeoutError
            await asyncio.wait_for(event.wait(), timeout=budget)
        except asyncio.TimeoutError:
            await self._expire(request_id)
            raise ApprovalTimeout(
                f"Nobody decided approval {request_id} in time; {req.tool} did not run",
                details={"id": request_id, "tool": req.tool},
            ) from None
        finally:
            async with self._lock:
                self._waiters.pop(request_id, None)
        async with self._lock:
            settled = self._requests.get(request_id, req)
        return self._settled(settled)

    def _settled(self, req: ApprovalRequest) -> ApprovalRequest:
        if req.state == ApprovalState.APPROVED:
            return req
        if req.state == ApprovalState.DENIED:
            raise ApprovalDenied(
                f"{req.tool} was denied" + (f": {req.note}" if req.note else ""),
                details={"id": req.id, "tool": req.tool, "by": req.decided_by},
            )
        raise ApprovalTimeout(
            f"Approval {req.id} expired; {req.tool} did not run",
            details={"id": req.id, "tool": req.tool},
        )

    async def decide(
        self,
        request_id: str,
        *,
        approved: bool,
        by: Optional[str] = None,
        note: Optional[str] = None,
        arguments: Optional[Dict[str, Any]] = None,
    ) -> Optional[ApprovalRequest]:
        """Settle a request. Returns it, or ``None`` when this gate never had it
        (so a fan-out decide can ask every node and only the holder acts).

        ``arguments`` replaces what the model proposed — an approver may correct a
        value rather than reject the call outright. First decision wins; a second
        one returns the already-settled request untouched.
        """
        await self.load()
        async with self._lock:
            req = self._requests.get(request_id)
            if req is None:
                return None
            if req.state != ApprovalState.PENDING:
                return req
            if req.is_expired:
                req.state = ApprovalState.EXPIRED
            else:
                req.state = ApprovalState.APPROVED if approved else ApprovalState.DENIED
                if approved and arguments is not None:
                    req.arguments = dict(arguments)
            req.decided_by = by
            req.note = note
            req.decided_at = time.time()
            event = self._waiters.get(request_id)
        await self._write(req)
        _logger.info(
            "Approval %s %s by %s", req.id, req.state.value, by or "?",
            extra={"approval_id": req.id, "tool": req.tool, "state": req.state.value},
        )
        if event is not None:
            event.set()
        return req

    async def approve(self, request_id: str, **kwargs: Any) -> Optional[ApprovalRequest]:
        return await self.decide(request_id, approved=True, **kwargs)

    async def deny(self, request_id: str, **kwargs: Any) -> Optional[ApprovalRequest]:
        return await self.decide(request_id, approved=False, **kwargs)

    async def _expire(self, request_id: str) -> None:
        async with self._lock:
            req = self._requests.get(request_id)
            if req is None or req.state != ApprovalState.PENDING:
                return
            req.state = ApprovalState.EXPIRED
            req.decided_at = time.time()
        await self._write(req)

    # --- reads ---

    async def get(self, request_id: str) -> Optional[ApprovalRequest]:
        await self.load()
        async with self._lock:
            return self._requests.get(request_id)

    async def pending(self) -> List[dict]:
        """Every request still awaiting a decision, oldest first.

        Requests past their deadline are settled as ``expired`` here rather than
        being offered to an operator who can no longer affect the outcome.
        """
        await self.load()
        async with self._lock:
            stale = [
                r for r in self._requests.values()
                if r.state == ApprovalState.PENDING and r.is_expired
            ]
            live = [
                r.to_dict() for r in self._requests.values()
                if r.state == ApprovalState.PENDING and not r.is_expired
            ]
        for req in stale:
            await self._expire(req.id)
        live.sort(key=lambda r: r["requested_at"])
        return live

    async def gate(
        self,
        tool: str,
        arguments: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """File a request, wait for it, and return the arguments cleared to run.

        Raises :class:`ApprovalDenied` or :class:`ApprovalTimeout` otherwise, so
        a caller that gets a return value can act::

            args = await gate.gate("billing-refund", {"order_id": "o1"})
            await tool.call(args)
        """
        req = await self.request(tool, arguments, **kwargs)
        return (await self.wait(req.id)).arguments

decide_key property

Where this node accepts decisions.

key property

Where this node lists its pending requests.

decide(request_id, *, approved, by=None, note=None, arguments=None) async

Settle a request. Returns it, or None when this gate never had it (so a fan-out decide can ask every node and only the holder acts).

arguments replaces what the model proposed — an approver may correct a value rather than reject the call outright. First decision wins; a second one returns the already-settled request untouched.

Source code in src/istos/agent/approval.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
async def decide(
    self,
    request_id: str,
    *,
    approved: bool,
    by: Optional[str] = None,
    note: Optional[str] = None,
    arguments: Optional[Dict[str, Any]] = None,
) -> Optional[ApprovalRequest]:
    """Settle a request. Returns it, or ``None`` when this gate never had it
    (so a fan-out decide can ask every node and only the holder acts).

    ``arguments`` replaces what the model proposed — an approver may correct a
    value rather than reject the call outright. First decision wins; a second
    one returns the already-settled request untouched.
    """
    await self.load()
    async with self._lock:
        req = self._requests.get(request_id)
        if req is None:
            return None
        if req.state != ApprovalState.PENDING:
            return req
        if req.is_expired:
            req.state = ApprovalState.EXPIRED
        else:
            req.state = ApprovalState.APPROVED if approved else ApprovalState.DENIED
            if approved and arguments is not None:
                req.arguments = dict(arguments)
        req.decided_by = by
        req.note = note
        req.decided_at = time.time()
        event = self._waiters.get(request_id)
    await self._write(req)
    _logger.info(
        "Approval %s %s by %s", req.id, req.state.value, by or "?",
        extra={"approval_id": req.id, "tool": req.tool, "state": req.state.value},
    )
    if event is not None:
        event.set()
    return req

gate(tool, arguments=None, **kwargs) async

File a request, wait for it, and return the arguments cleared to run.

Raises :class:ApprovalDenied or :class:ApprovalTimeout otherwise, so a caller that gets a return value can act::

args = await gate.gate("billing-refund", {"order_id": "o1"})
await tool.call(args)
Source code in src/istos/agent/approval.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
async def gate(
    self,
    tool: str,
    arguments: Optional[Dict[str, Any]] = None,
    **kwargs: Any,
) -> Dict[str, Any]:
    """File a request, wait for it, and return the arguments cleared to run.

    Raises :class:`ApprovalDenied` or :class:`ApprovalTimeout` otherwise, so
    a caller that gets a return value can act::

        args = await gate.gate("billing-refund", {"order_id": "o1"})
        await tool.call(args)
    """
    req = await self.request(tool, arguments, **kwargs)
    return (await self.wait(req.id)).arguments

load() async

Recover requests still pending from a previous run. Runs once.

Source code in src/istos/agent/approval.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def load(self) -> None:
    """Recover requests still pending from a previous run. Runs once."""
    if self._loaded:
        return
    self._loaded = True
    if self._storage is None:
        return
    try:
        for request_id in await self._storage.get(self._index_key()) or []:
            raw = await self._storage.get(self._req_key(request_id))
            if isinstance(raw, dict) and request_id not in self._requests:
                self._requests[request_id] = ApprovalRequest.from_dict(raw)
    except Exception:  # recovery is best-effort — never block the gate
        _logger.exception("Could not recover pending approvals")

pending() async

Every request still awaiting a decision, oldest first.

Requests past their deadline are settled as expired here rather than being offered to an operator who can no longer affect the outcome.

Source code in src/istos/agent/approval.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
async def pending(self) -> List[dict]:
    """Every request still awaiting a decision, oldest first.

    Requests past their deadline are settled as ``expired`` here rather than
    being offered to an operator who can no longer affect the outcome.
    """
    await self.load()
    async with self._lock:
        stale = [
            r for r in self._requests.values()
            if r.state == ApprovalState.PENDING and r.is_expired
        ]
        live = [
            r.to_dict() for r in self._requests.values()
            if r.state == ApprovalState.PENDING and not r.is_expired
        ]
    for req in stale:
        await self._expire(req.id)
    live.sort(key=lambda r: r["requested_at"])
    return live

request(tool, arguments=None, *, prefix=None, reason=None, requester=None, conversation_id=None, timeout_s=-1.0) async

File a request and return it, without waiting. timeout_s defaults to the gate's; pass None for no deadline.

Source code in src/istos/agent/approval.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
async def request(
    self,
    tool: str,
    arguments: Optional[Dict[str, Any]] = None,
    *,
    prefix: Optional[str] = None,
    reason: Optional[str] = None,
    requester: Optional[str] = None,
    conversation_id: Optional[str] = None,
    timeout_s: Optional[float] = -1.0,
) -> ApprovalRequest:
    """File a request and return it, without waiting. ``timeout_s`` defaults
    to the gate's; pass ``None`` for no deadline."""
    await self.load()
    ttl = self.timeout_s if timeout_s == -1.0 else timeout_s
    req = ApprovalRequest(
        id=uuid.uuid4().hex,
        tool=tool,
        arguments=dict(arguments or {}),
        prefix=prefix,
        reason=reason,
        requester=requester or self._service,
        conversation_id=conversation_id,
        expires_at=time.time() + ttl if ttl else 0.0,
    )
    async with self._lock:
        self._requests[req.id] = req
        self._waiters[req.id] = asyncio.Event()
    await self._write(req)
    _logger.info(
        "Approval %s pending for tool %s", req.id, tool,
        extra={"approval_id": req.id, "tool": tool, "conversation_id": conversation_id},
    )
    if self.on_request is not None:
        try:
            result = self.on_request(req)
            if asyncio.iscoroutine(result):
                await result
        except Exception:  # a broken notifier must not strand the request
            _logger.exception("approval on_request callback failed")
    return req

wait(request_id, *, timeout_s=-1.0) async

Block until the request is settled.

Returns the approved request (its arguments are what to run — an approver may have edited them). Raises :class:ApprovalDenied on a no, :class:ApprovalTimeout when the deadline passes, and :class:NotFoundError for an unknown id.

Source code in src/istos/agent/approval.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
async def wait(
    self, request_id: str, *, timeout_s: Optional[float] = -1.0,
) -> ApprovalRequest:
    """Block until the request is settled.

    Returns the approved request (its ``arguments`` are what to run — an
    approver may have edited them). Raises :class:`ApprovalDenied` on a no,
    :class:`ApprovalTimeout` when the deadline passes, and
    :class:`NotFoundError` for an unknown id.
    """
    async with self._lock:
        req = self._requests.get(request_id)
        event = self._waiters.get(request_id)
    if req is None:
        raise NotFoundError(f"No approval request {request_id!r}")
    if req.state != ApprovalState.PENDING:
        return self._settled(req)

    # The gate's own deadline and the caller's, whichever comes first.
    budget = self.timeout_s if timeout_s == -1.0 else timeout_s
    if req.expires_at:
        remaining = req.expires_at - time.time()
        budget = remaining if budget is None else min(budget, remaining)
    if event is None:  # recovered from storage: nobody in this process waits
        raise ApprovalTimeout(
            f"Approval {request_id} was filed by an earlier run of this node",
            details={"id": request_id},
        )
    try:
        if budget is not None and budget <= 0:
            raise asyncio.TimeoutError
        await asyncio.wait_for(event.wait(), timeout=budget)
    except asyncio.TimeoutError:
        await self._expire(request_id)
        raise ApprovalTimeout(
            f"Nobody decided approval {request_id} in time; {req.tool} did not run",
            details={"id": request_id, "tool": req.tool},
        ) from None
    finally:
        async with self._lock:
            self._waiters.pop(request_id, None)
    async with self._lock:
        settled = self._requests.get(request_id, req)
    return self._settled(settled)

ApprovalRequest dataclass

One pending (or settled) request for a human decision.

Source code in src/istos/agent/approval.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass
class ApprovalRequest:
    """One pending (or settled) request for a human decision."""

    id: str
    tool: str
    arguments: Dict[str, Any] = field(default_factory=dict)
    prefix: Optional[str] = None
    state: ApprovalState = ApprovalState.PENDING
    requested_at: float = field(default_factory=time.time)
    expires_at: float = 0.0            # 0 → no deadline
    reason: Optional[str] = None       # why this tool needs a human
    requester: Optional[str] = None    # which service asked
    conversation_id: Optional[str] = None
    decided_by: Optional[str] = None
    decided_at: float = 0.0
    note: Optional[str] = None         # the human's comment, shown to the model

    def to_dict(self) -> dict:
        d = asdict(self)
        d["state"] = self.state.value
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "ApprovalRequest":
        data = dict(d)
        data["state"] = ApprovalState(data.get("state", "pending"))
        known = {f for f in cls.__dataclass_fields__}
        return cls(**{k: v for k, v in data.items() if k in known})

    @property
    def is_expired(self) -> bool:
        return self.expires_at > 0 and time.time() >= self.expires_at

ApprovalTimeout

Bases: IstosError

Nobody decided before the deadline, so the call did not happen.

Source code in src/istos/agent/approval.py 
60
61
62
63
64
class ApprovalTimeout(IstosError):
    """Nobody decided before the deadline, so the call did not happen."""

    def __init__(self, message: str = "Approval timed out", **kwargs: Any):
        super().__init__(message, code="approval_timeout", status=504, **kwargs)

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"]})

approval=True (or a string reason) makes the loop stop for a human before the call — see :mod:istos.agent.approval. The tool's owner can declare it instead, with @handle(approval=True), in which case discovery carries it here on its own.

Source code in src/istos/agent/tools.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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"]})

    ``approval=True`` (or a string reason) makes the loop stop for a human before
    the call — see :mod:`istos.agent.approval`. The tool's owner can declare it
    instead, with ``@handle(approval=True)``, in which case discovery carries it
    here on its own.
    """

    def __init__(
        self,
        prefix: str,
        *,
        app: Any = None,
        name: Optional[str] = None,
        description: str = "",
        parameters: Optional[dict] = None,
        invoke: Optional[Callable[..., Awaitable[Any]]] = None,
        approval: Union[bool, str] = False,
    ) -> 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.approval = approval
        self._app = app
        self._invoke = invoke

    @property
    def requires_approval(self) -> bool:
        return bool(self.approval)

    @property
    def approval_reason(self) -> Optional[str]:
        """The reason an approver is shown, when the flag carried one."""
        return self.approval if isinstance(self.approval, str) else None

    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
        )

approval_reason property

The reason an approver is shown, when the flag carried one.

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 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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 
75
76
77
78
79
80
81
82
83
84
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,
        },
    }

Model

Bases: Protocol

Source code in src/istos/agent/model.py 
48
49
50
51
52
53
54
55
56
57
@runtime_checkable
class Model(Protocol):
    async def complete(
        self,
        messages: List[dict],
        *,
        tools: Optional[List[dict]] = None,
    ) -> ModelReply:
        """Next assistant turn. ``tools`` is the OpenAI tools array, or None."""
        ...

complete(messages, *, tools=None) async

Next assistant turn. tools is the OpenAI tools array, or None.

Source code in src/istos/agent/model.py 
50
51
52
53
54
55
56
57
async def complete(
    self,
    messages: List[dict],
    *,
    tools: Optional[List[dict]] = None,
) -> ModelReply:
    """Next assistant turn. ``tools`` is the OpenAI tools array, or None."""
    ...

ModelError

Bases: RuntimeError

The model was unreachable, or answered with something unusable.

Source code in src/istos/agent/model.py 
60
61
class ModelError(RuntimeError):
    """The model was unreachable, or answered with something unusable."""

ModelReply dataclass

What the model returned for one completion turn.

model, finish_reason, and usage ({prompt_tokens, completion_tokens, …}) are optional telemetry an adapter may fill in; the loop puts them on its completion span. A custom model can leave them unset.

Source code in src/istos/agent/model.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass
class ModelReply:
    """What the model returned for one completion turn.

    ``model``, ``finish_reason``, and ``usage`` (``{prompt_tokens,
    completion_tokens, …}``) are optional telemetry an adapter may fill in; the
    loop puts them on its completion span. A custom model can leave them unset.
    """

    content: Optional[str] = None
    tool_calls: List[ToolCall] = field(default_factory=list)
    model: Optional[str] = None
    finish_reason: Optional[str] = None
    usage: Optional[Dict[str, Any]] = None

OpenAIChatModel

Thin /v1/chat/completions client (non-streaming, with tool calls).

Uses aiohttp — already an Istos dependency — so no extra install for the common OpenAI-compatible local servers::

model = OpenAIChatModel(
    base_url="http://127.0.0.1:1234/v1",
    model="qwen/qwen3.5-9b",
)

One connection pool is shared across :meth:complete calls, so an agent loop's steps reuse a live connection instead of reconnecting per turn. The pool opens on first use and closes with :meth:aclose; use the model as an async context manager when its lifetime is scoped::

async with OpenAIChatModel(base_url=…, model=…) as model:
    await run_agent(model, tools, messages)

Long-lived models (built once at import, used by a handler for the process lifetime) can skip the close — the pool goes away with the loop.

Source code in src/istos/agent/model.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
class OpenAIChatModel:
    """Thin ``/v1/chat/completions`` client (non-streaming, with tool calls).

    Uses aiohttp — already an Istos dependency — so no extra install for the
    common OpenAI-compatible local servers::

        model = OpenAIChatModel(
            base_url="http://127.0.0.1:1234/v1",
            model="qwen/qwen3.5-9b",
        )

    One connection pool is shared across :meth:`complete` calls, so an agent
    loop's steps reuse a live connection instead of reconnecting per turn. The
    pool opens on first use and closes with :meth:`aclose`; use the model as an
    async context manager when its lifetime is scoped::

        async with OpenAIChatModel(base_url=…, model=…) as model:
            await run_agent(model, tools, messages)

    Long-lived models (built once at import, used by a handler for the process
    lifetime) can skip the close — the pool goes away with the loop.
    """

    def __init__(
        self,
        *,
        base_url: str,
        model: str,
        api_key: Optional[str] = None,
        timeout_s: float = 120.0,
        temperature: float = 0.0,
        max_tokens: Optional[int] = None,
        extra_body: Optional[Dict[str, Any]] = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.model = model
        self.api_key = api_key
        self.timeout_s = timeout_s
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.extra_body = extra_body or {}
        self._session: Optional[aiohttp.ClientSession] = None
        self._session_loop: Optional[asyncio.AbstractEventLoop] = None

    async def _get_session(self) -> aiohttp.ClientSession:
        """The shared session, created on first use.

        A ``ClientSession`` is bound to the loop that created it, so a model
        reused across loops (a test suite, ``asyncio.run`` called twice) gets a
        fresh session rather than a connector wired to a dead loop.
        """
        loop = asyncio.get_running_loop()
        if self._session is not None and not self._session.closed:
            if self._session_loop is loop:
                return self._session
            # Previous loop is gone; its transports cannot be closed from here.
            self._session = None
        self._session = aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=self.timeout_s)
        )
        self._session_loop = loop
        return self._session

    async def aclose(self) -> None:
        """Close the shared connection pool. Safe to call more than once."""
        session, self._session = self._session, None
        self._session_loop = None
        if session is not None and not session.closed:
            await session.close()

    async def __aenter__(self) -> "OpenAIChatModel":
        return self

    async def __aexit__(self, *exc: Any) -> None:
        await self.aclose()

    async def complete(
        self,
        messages: List[dict],
        *,
        tools: Optional[List[dict]] = None,
    ) -> ModelReply:
        body: Dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "temperature": self.temperature,
            **self.extra_body,
        }
        if self.max_tokens is not None:
            body["max_tokens"] = self.max_tokens
        if tools:
            body["tools"] = tools
            body["tool_choice"] = "auto"

        headers: Dict[str, str] = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        session = await self._get_session()
        try:
            async with session.post(
                f"{self.base_url}/chat/completions",
                json=body,
                headers=headers,
            ) as resp:
                if resp.status != 200:
                    detail = await resp.text()
                    raise ModelError(
                        f"chat/completions returned {resp.status}: {detail[:400]}"
                    )
                payload = await resp.json()
        except aiohttp.ClientError as exc:
            raise ModelError(
                f"Could not reach model at {self.base_url}: {exc}"
            ) from exc

        try:
            message = payload["choices"][0]["message"]
        except (KeyError, IndexError, TypeError) as exc:
            raise ModelError(
                f"Unexpected response shape: {json.dumps(payload)[:400]}"
            ) from exc

        finish = payload["choices"][0].get("finish_reason")
        usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else None

        content = message.get("content")
        if isinstance(content, str):
            content = content.strip() or None
        else:
            content = None

        tool_calls: List[ToolCall] = []
        for raw in message.get("tool_calls") or []:
            try:
                fn = raw["function"]
                args_raw = fn.get("arguments") or "{}"
                if isinstance(args_raw, str):
                    arguments = json.loads(args_raw) if args_raw.strip() else {}
                elif isinstance(args_raw, dict):
                    arguments = args_raw
                else:
                    arguments = {}
                if not isinstance(arguments, dict):
                    arguments = {}
                tool_calls.append(
                    ToolCall(
                        id=str(raw.get("id") or uuid.uuid4().hex),
                        name=str(fn["name"]),
                        arguments=arguments,
                    )
                )
            except (KeyError, TypeError, json.JSONDecodeError) as exc:
                _logger.warning(
                    "Skipping malformed tool_call from model: %s", exc,
                    extra={"raw": raw},
                )
                continue

        return ModelReply(
            content=content,
            tool_calls=tool_calls,
            model=payload.get("model") or self.model,
            finish_reason=finish,
            usage=usage,
        )

aclose() async

Close the shared connection pool. Safe to call more than once.

Source code in src/istos/agent/model.py 
127
128
129
130
131
132
async def aclose(self) -> None:
    """Close the shared connection pool. Safe to call more than once."""
    session, self._session = self._session, None
    self._session_loop = None
    if session is not None and not session.closed:
        await session.close()

ToolCall dataclass

One function call the model asked for.

Source code in src/istos/agent/model.py 
23
24
25
26
27
28
29
@dataclass
class ToolCall:
    """One function call the model asked for."""

    id: str
    name: str
    arguments: Dict[str, Any]

build_registry(entry)

Every agent reachable from entry through handoffs, keyed by name.

Source code in src/istos/agent/multi.py 
 98
 99
100
101
102
103
104
105
106
107
108
def build_registry(entry: Agent) -> Dict[str, Agent]:
    """Every agent reachable from ``entry`` through ``handoffs``, keyed by name."""
    registry: Dict[str, Agent] = {}
    stack = [entry]
    while stack:
        agent = stack.pop()
        if agent.name in registry:
            continue
        registry[agent.name] = agent
        stack.extend(agent.handoffs)
    return registry

decide_approval(app, request_id, *, approved, by=None, note=None, timeout_s=3.0, **query_kwargs) async

Settle a request from anywhere on the mesh, without knowing which node holds it.

Asks every .istos/approvals/*/decide key and returns the settled request from the one node that had it::

await decide_approval(app, req["id"], approved=False, by="amir",
                      note="wrong order")

Raises :class:~istos.errors.NotFoundError when no node recognised the id — it was already settled, it expired, or the waiting node is gone.

Source code in src/istos/agent/approval.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
async def decide_approval(
    app: Any,
    request_id: str,
    *,
    approved: bool,
    by: Optional[str] = None,
    note: Optional[str] = None,
    timeout_s: float = 3.0,
    **query_kwargs: Any,
) -> dict:
    """Settle a request from anywhere on the mesh, without knowing which node holds it.

    Asks every ``.istos/approvals/*/decide`` key and returns the settled request
    from the one node that had it::

        await decide_approval(app, req["id"], approved=False, by="amir",
                              note="wrong order")

    Raises :class:`~istos.errors.NotFoundError` when no node recognised the id —
    it was already settled, it expired, or the waiting node is gone.
    """
    replies = await app.query_once(
        DECIDE_WILDCARD,
        request_id=request_id, approved=approved, by=by, note=note,
        timeout_s=timeout_s, consolidate_replies=False, **query_kwargs,
    )
    if replies is None:
        replies = []
    if not isinstance(replies, list):
        replies = [replies]
    for reply in replies:
        if isinstance(reply, dict) and reply.get("matched"):
            request: dict = reply.get("request") or {}
            return request
    raise NotFoundError(
        f"No node is holding approval {request_id!r}",
        details={"id": request_id},
    )

drive_agents(session, entry, *, max_steps=8, max_messages=40, token=None, timeout_s=5.0, send_events=True, approvals=None) async

Channel helper: reload history, then run :func:run_multi_agent per turn.

The active agent persists across turns within a session, and is restored on reconnect from persisted handoff frames (send_events=True); otherwise a resumed session restarts at entry. token forwards to tool calls under whichever agent is active. See :func:~istos.agent.loop.drive_channel for the send_events payload shapes and for how approvals behaves.

Source code in src/istos/agent/multi.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
async def drive_agents(
    session: ChannelSession,
    entry: Agent,
    *,
    max_steps: int = 8,
    max_messages: Optional[int] = 40,
    token: Optional[Union[bytes, str]] = None,
    timeout_s: float = 5.0,
    send_events: bool = True,
    approvals: Optional["ApprovalGate"] = None,
) -> None:
    """Channel helper: reload history, then run :func:`run_multi_agent` per turn.

    The active agent persists across turns within a session, and is restored on
    reconnect from persisted ``handoff`` frames (``send_events=True``); otherwise
    a resumed session restarts at ``entry``. ``token`` forwards to tool calls
    under whichever agent is active. See :func:`~istos.agent.loop.drive_channel`
    for the ``send_events`` payload shapes and for how ``approvals`` behaves.
    """
    registry = build_registry(entry)
    history = await session.history()
    messages = history_to_messages(history)
    active = _active_from_history(history, registry, entry)

    async for msg in session:
        messages.append({"role": "user", "content": user_text(msg)})
        async for event in run_multi_agent(
            active, messages,
            max_steps=max_steps, max_messages=max_messages,
            token=token, timeout_s=timeout_s,
            approvals=approvals, conversation_id=session.conversation_id,
        ):
            if event.kind == "handoff" and event.name is not None:
                target = registry.get(event.name)
                if target is not None:
                    active = target
            if event.kind == "done":
                continue
            if send_events:
                frame = {
                    "kind": event.kind,
                    "content": event.content,
                    "name": event.name,
                    "arguments": event.arguments,
                    "tool_call_id": event.tool_call_id,
                    "error": event.error,
                }
                if event.approval_id is not None:
                    frame["approval_id"] = event.approval_id
                await session.send(frame)
            elif event.kind == "message" and event.content is not None:
                await session.send(event.content)

drive_channel(session, model, tools, *, system=None, max_steps=8, max_messages=40, token=None, timeout_s=5.0, send_events=True, approvals=None) async

Channel helper: reload history, then run :func:run_agent per inbound turn.

By default each :class:AgentEvent is sent as a dict {"kind", "content", …}. Set send_events=False to send only the final message content (plain string). max_messages bounds the reused log so a long-lived session does not grow unboundedly; pass None to disable.

With approvals set, a gated tool sends an approval_request frame (carrying approval_id) and the turn pauses. The decision comes back over the fabric, not this socket — decide_approval(app, approval_id, …), or the HTTP gateway in front of it — so an operator who is not this channel's peer can answer, and a peer cannot approve merely by holding the socket.

Source code in src/istos/agent/loop.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
async def drive_channel(
    session: ChannelSession,
    model: Model,
    tools: Sequence[MeshTool],
    *,
    system: Optional[str] = None,
    max_steps: int = 8,
    max_messages: Optional[int] = 40,
    token: Optional[Union[bytes, str]] = None,
    timeout_s: float = 5.0,
    send_events: bool = True,
    approvals: Optional["ApprovalGate"] = None,
) -> None:
    """Channel helper: reload history, then run :func:`run_agent` per inbound turn.

    By default each :class:`AgentEvent` is sent as a dict
    ``{"kind", "content", …}``. Set ``send_events=False`` to send only the final
    ``message`` content (plain string). ``max_messages`` bounds the reused log so
    a long-lived session does not grow unboundedly; pass ``None`` to disable.

    With ``approvals`` set, a gated tool sends an ``approval_request`` frame
    (carrying ``approval_id``) and the turn pauses. The decision comes back over
    the fabric, not this socket — ``decide_approval(app, approval_id, …)``, or the
    HTTP gateway in front of it — so an operator who is not this channel's peer
    can answer, and a peer cannot approve merely by holding the socket.
    """
    history = await session.history()
    messages = history_to_messages(history, system=system)
    if system and not any(m.get("role") == "system" for m in messages):
        messages.insert(0, {"role": "system", "content": system})

    async for msg in session:
        messages.append({"role": "user", "content": user_text(msg)})
        async for event in run_agent(
            model, tools, messages,
            max_steps=max_steps, max_messages=max_messages,
            token=token, timeout_s=timeout_s,
            approvals=approvals,
            conversation_id=session.conversation_id,
        ):
            if event.kind == "done":
                continue
            if send_events:
                frame = {
                    "kind": event.kind,
                    "content": event.content,
                    "name": event.name,
                    "arguments": event.arguments,
                    "tool_call_id": event.tool_call_id,
                    "error": event.error,
                }
                if event.approval_id is not None:
                    frame["approval_id"] = event.approval_id
                await session.send(frame)
            elif event.kind == "message" and event.content is not None:
                await session.send(event.content)

history_to_messages(history, *, system=None, include_tools=True)

Map a durable channel log ([{dir, data, ts}, …]) into chat messages.

With include_tools (the default) the tool transcript is reconstructed: each persisted tool_call frame becomes an assistant tool_calls message followed by the tool message carrying its result, so a reconnecting agent sees the same context it had live. A tool_call with no matching tool_result in the log (a crash mid-tool) is dropped so the sequence stays valid. Set include_tools=False to rebuild plain text only.

Source code in src/istos/agent/loop.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def history_to_messages(
    history: Sequence[dict],
    *,
    system: Optional[str] = None,
    include_tools: bool = True,
) -> List[dict]:
    """Map a durable channel log (``[{dir, data, ts}, …]``) into chat messages.

    With ``include_tools`` (the default) the tool transcript is reconstructed:
    each persisted ``tool_call`` frame becomes an assistant ``tool_calls`` message
    followed by the ``tool`` message carrying its result, so a reconnecting agent
    sees the same context it had live. A ``tool_call`` with no matching
    ``tool_result`` in the log (a crash mid-tool) is dropped so the sequence stays
    valid. Set ``include_tools=False`` to rebuild plain text only.
    """
    results_by_id: Dict[str, dict] = {}
    if include_tools:
        for turn in history:
            if turn.get("dir") != "out":
                continue
            data = turn.get("data")
            if isinstance(data, dict) and data.get("kind") == "tool_result":
                tid = data.get("tool_call_id")
                if tid is not None:
                    results_by_id[tid] = data

    messages: List[dict] = []
    if system:
        messages.append({"role": "system", "content": system})
    for turn in history:
        data = turn.get("data")
        direction = turn.get("dir")
        if direction == "in":
            messages.append({"role": "user", "content": user_text(data)})
        elif direction == "out":
            if not isinstance(data, dict):
                if user_text(data):
                    messages.append({"role": "assistant", "content": user_text(data)})
                continue
            kind = data.get("kind")
            if kind == "tool_call":
                if not include_tools:
                    continue
                tid = data.get("tool_call_id")
                result = results_by_id.get(tid) if isinstance(tid, str) else None
                if result is None:
                    continue
                messages.append(_tool_call_message(data))
                messages.append({
                    "role": "tool",
                    "tool_call_id": tid,
                    "content": str(result.get("content") or ""),
                })
                continue
            if kind in ("tool_result", "done", "handoff"):
                continue
            text = data.get("content") if kind == "message" else user_text(data)
            if text:
                messages.append({"role": "assistant", "content": str(text)})
    return messages

list_approvals(app, *, timeout_s=3.0, **query_kwargs) async

Every pending approval on the fabric, from any node.

Asks .istos/approvals/*: one key per waiting node, so all of them answer::

for req in await list_approvals(app):
    print(req["id"], req["tool"], req["arguments"])
Source code in src/istos/agent/approval.py 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
async def list_approvals(app: Any, *, timeout_s: float = 3.0, **query_kwargs: Any) -> List[dict]:
    """Every pending approval on the fabric, from any node.

    Asks ``.istos/approvals/*``: one key per waiting node, so all of them answer::

        for req in await list_approvals(app):
            print(req["id"], req["tool"], req["arguments"])
    """
    replies = await app.query_once(
        APPROVALS_WILDCARD, timeout_s=timeout_s, consolidate_replies=False, **query_kwargs
    )
    if replies is None:
        return []
    if not isinstance(replies, list):
        replies = [replies]
    out: List[dict] = []
    for reply in replies:
        if not isinstance(reply, dict) or is_error_payload(reply):
            continue
        for req in reply.get("pending") or []:
            if isinstance(req, dict):
                out.append({**req, "service": reply.get("service")})
    out.sort(key=lambda r: r.get("requested_at") or 0)
    return out

run_agent(model, tools, messages, *, max_steps=8, max_messages=None, token=None, timeout_s=5.0, approvals=None, conversation_id=None) async

Run plan → tool → observe until the model returns text or max_steps.

Mutates messages in place so the caller can keep a multi-turn conversation. Each mesh tool call forwards token on query_once. Pass max_messages to bound the log before each completion (system prompt kept).

approvals is an :class:~istos.agent.approval.ApprovalGate (usually app.approvals()). A tool marked approval=True suspends the loop there: an approval_request event goes out, the gate waits for a human, and the call runs only if they said yes. A refusal or a timeout becomes a failed tool_result the model can respond to — the turn continues, the tool does not run. Passing tools that need approval without a gate is an error, so a missing gate can never read as a pass.

Source code in src/istos/agent/loop.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
async def run_agent(
    model: Model,
    tools: Sequence[MeshTool],
    messages: List[dict],
    *,
    max_steps: int = 8,
    max_messages: Optional[int] = None,
    token: Optional[Union[bytes, str]] = None,
    timeout_s: float = 5.0,
    approvals: Optional["ApprovalGate"] = None,
    conversation_id: Optional[str] = None,
) -> AsyncIterator[AgentEvent]:
    """Run plan → tool → observe until the model returns text or ``max_steps``.

    Mutates ``messages`` in place so the caller can keep a multi-turn
    conversation. Each mesh tool call forwards ``token`` on ``query_once``. Pass
    ``max_messages`` to bound the log before each completion (system prompt kept).

    ``approvals`` is an :class:`~istos.agent.approval.ApprovalGate` (usually
    ``app.approvals()``). A tool marked ``approval=True`` suspends the loop there:
    an ``approval_request`` event goes out, the gate waits for a human, and the
    call runs only if they said yes. A refusal or a timeout becomes a failed
    ``tool_result`` the model can respond to — the turn continues, the tool does
    not run. Passing tools that need approval without a gate is an error, so a
    missing gate can never read as a pass.
    """
    if max_steps < 1:
        raise ValueError("max_steps must be >= 1")

    if approvals is None:
        gated = [t.name for t in tools if t.requires_approval]
        if gated:
            raise ValueError(
                f"Tools {gated} require human approval but no gate was passed. "
                "Pass approvals=app.approvals(), or drop approval= from the tool."
            )

    catalog = _by_name(tools)
    schemas = [t.openai_schema() for t in tools] or None

    for step in range(max_steps):
        if max_messages is not None:
            _trim_messages(messages, max_messages)
        with span(
            "istos.agent.completion",
            {
                "gen_ai.operation.name": "chat",
                "istos.agent.step": step + 1,
                "istos.agent.messages": len(messages),
                "istos.agent.tools": len(tools),
            },
        ) as sp:
            reply = await model.complete(messages, tools=schemas)
            set_span_attributes(sp, _completion_attrs(reply))
        messages.append(_assistant_message(reply))

        if not reply.tool_calls:
            if reply.content:
                yield AgentEvent(kind="message", content=reply.content)
            else:
                # No text and no usable tool call — never end a turn silently.
                yield AgentEvent(
                    kind="message",
                    content="The model returned an empty response.",
                    error=True,
                )
            yield AgentEvent(kind="done")
            return

        for tc in reply.tool_calls:
            async for event in _dispatch_tool(
                catalog, tc, messages,
                token=token, timeout_s=timeout_s,
                approvals=approvals, conversation_id=conversation_id,
            ):
                yield event

        _logger.debug(
            "Agent step %s finished %s tool call(s)",
            step + 1, len(reply.tool_calls),
            extra={"step": step + 1, "n_tools": len(reply.tool_calls)},
        )

    # Hit the ceiling while still tool-calling — surface what we have.
    yield AgentEvent(
        kind="message",
        content=f"Stopped after {max_steps} tool steps without a final answer.",
    )
    yield AgentEvent(kind="done")

run_multi_agent(active, messages, *, max_steps=8, max_messages=None, token=None, timeout_s=5.0, approvals=None, conversation_id=None) async

Run the loop starting at active, switching agents on handoff.

Mutates messages in place (shared history). Emits the same events as :func:~istos.agent.loop.run_agent plus handoff when the active agent changes; the last handoff event names the agent that should drive the next turn. token is forwarded to every tool call, across handoffs.

approvals gates tools marked approval=True on a human, wherever in the handoff graph they are reached — see :func:~istos.agent.loop.run_agent.

Source code in src/istos/agent/multi.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
async def run_multi_agent(
    active: Agent,
    messages: List[dict],
    *,
    max_steps: int = 8,
    max_messages: Optional[int] = None,
    token: Optional[Union[bytes, str]] = None,
    timeout_s: float = 5.0,
    approvals: Optional["ApprovalGate"] = None,
    conversation_id: Optional[str] = None,
) -> AsyncIterator[AgentEvent]:
    """Run the loop starting at ``active``, switching agents on handoff.

    Mutates ``messages`` in place (shared history). Emits the same events as
    :func:`~istos.agent.loop.run_agent` plus ``handoff`` when the active agent
    changes; the last ``handoff`` event names the agent that should drive the
    next turn. ``token`` is forwarded to every tool call, across handoffs.

    ``approvals`` gates tools marked ``approval=True`` on a human, wherever in the
    handoff graph they are reached — see :func:`~istos.agent.loop.run_agent`.
    """
    if max_steps < 1:
        raise ValueError("max_steps must be >= 1")

    if approvals is None:
        gated = [
            t.name
            for agent in build_registry(active).values()
            for t in agent.tools
            if t.requires_approval
        ]
        if gated:
            raise ValueError(
                f"Tools {gated} require human approval but no gate was passed. "
                "Pass approvals=app.approvals(), or drop approval= from the tool."
            )

    for step in range(max_steps):
        if max_messages is not None:
            _trim_messages(messages, max_messages)

        handoff_by_tool = {_transfer_tool_name(t): t for t in active.handoffs}
        catalog = _by_name(active.tools)
        schemas = [t.openai_schema() for t in active.tools]
        schemas += [_handoff_schema(t) for t in active.handoffs]

        with span(
            "istos.agent.completion",
            {
                "gen_ai.operation.name": "chat",
                "istos.agent.name": active.name,
                "istos.agent.step": step + 1,
                "istos.agent.messages": len(messages),
                "istos.agent.tools": len(active.tools),
                "istos.agent.handoffs": len(active.handoffs),
            },
        ) as sp:
            reply = await active.model.complete(
                _with_system(messages, active.system), tools=schemas or None
            )
            set_span_attributes(sp, _completion_attrs(reply))
        messages.append(_assistant_message(reply))

        if not reply.tool_calls:
            if reply.content:
                yield AgentEvent(kind="message", content=reply.content)
            else:
                yield AgentEvent(
                    kind="message",
                    content="The model returned an empty response.",
                    error=True,
                )
            yield AgentEvent(kind="done")
            return

        # Resolve every tool call against the agent that produced them, so each
        # gets a tool response (the API requires one per call). A handoff wins at
        # the end of the batch; the last transfer sets the next active agent.
        next_active = active
        for tc in reply.tool_calls:
            target = handoff_by_tool.get(tc.name)
            if target is not None:
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": f"Transferred to {target.name}.",
                })
                yield AgentEvent(
                    kind="handoff",
                    name=target.name,
                    content=f"{active.name} -> {target.name}",
                    tool_call_id=tc.id,
                )
                next_active = target
                continue

            async for event in _dispatch_tool(
                catalog, tc, messages,
                token=token, timeout_s=timeout_s,
                approvals=approvals, conversation_id=conversation_id,
            ):
                yield event

        if next_active is not active:
            _logger.debug(
                "Handoff %s -> %s", active.name, next_active.name,
                extra={"from": active.name, "to": next_active.name},
            )
            active = next_active

    yield AgentEvent(
        kind="message",
        content=f"Stopped after {max_steps} steps without a final answer.",
    )
    yield AgentEvent(kind="done")

tool_name(prefix)

The tool name for a key expression (or an agent name).

Every character outside [A-Za-z0-9_-] becomes -, so math/add is math-add. Distinct prefixes can collide (a/b and a.b both give a-b); pass an explicit name to :class:~istos.agent.MeshTool when that matters.

Source code in src/istos/discovery/naming.py 
24
25
26
27
28
29
30
31
32
def tool_name(prefix: str) -> str:
    """The tool name for a key expression (or an agent name).

    Every character outside ``[A-Za-z0-9_-]`` becomes ``-``, so ``math/add``
    is ``math-add``. Distinct prefixes can collide (``a/b`` and ``a.b`` both give
    ``a-b``); pass an explicit ``name`` to :class:`~istos.agent.MeshTool` when
    that matters.
    """
    return _UNSAFE_NAME.sub("-", prefix)

tools_from_discovery(app, *, services=None, prefixes=None, approval=None, timeout_s=3.0) async

Inventory the fabric and build tools for every remote @handle found.

The counterpart to :func:tools_from_handlers: instead of the local registry, this asks .istos/capabilities/* (via :meth:Istos.discover_capabilities), so a remote handler's schema and docstring come from the node that owns it rather than being written out by hand::

tools = await tools_from_discovery(app, services=["billing", "search"])
async for event in run_agent(model, tools, messages):
    ...

Pass services to whitelist by service name, prefixes to whitelist by key expression, approval to gate extra prefixes on a human (endpoints declaring @handle(approval=…) arrive gated already). Nodes with discovery disabled (Istos(enable_discovery=False)) do not answer and contribute nothing.

The catalogue is a snapshot: call again to pick up nodes that joined later. Duplicate prefixes across services collapse to the first one seen — the same key expression is the same endpoint either way.

Source code in src/istos/agent/tools.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def tools_from_discovery(
    app: Any,
    *,
    services: Optional[Sequence[str]] = None,
    prefixes: Optional[Sequence[str]] = None,
    approval: Optional[Sequence[str]] = None,
    timeout_s: float = 3.0,
) -> List[MeshTool]:
    """Inventory the fabric and build tools for every remote ``@handle`` found.

    The counterpart to :func:`tools_from_handlers`: instead of the local
    registry, this asks ``.istos/capabilities/*`` (via
    :meth:`Istos.discover_capabilities`), so a remote handler's schema and
    docstring come from the node that owns it rather than being written out by
    hand::

        tools = await tools_from_discovery(app, services=["billing", "search"])
        async for event in run_agent(model, tools, messages):
            ...

    Pass ``services`` to whitelist by service name, ``prefixes`` to whitelist by
    key expression, ``approval`` to gate extra prefixes on a human (endpoints
    declaring ``@handle(approval=…)`` arrive gated already). Nodes with discovery
    disabled (``Istos(enable_discovery=False)``) do not answer and contribute
    nothing.

    The catalogue is a snapshot: call again to pick up nodes that joined later.
    Duplicate prefixes across services collapse to the first one seen — the same
    key expression is the same endpoint either way.
    """
    manifests = await app.discover_capabilities(timeout_s=timeout_s)
    wanted = set(services) if services is not None else None
    out: List[MeshTool] = []
    seen: set = set()
    for service, manifest in sorted(manifests.items()):
        if wanted is not None and service not in wanted:
            continue
        for tool in tools_from_manifest(app, manifest, prefixes=prefixes, approval=approval):
            if tool.prefix in seen:
                continue
            seen.add(tool.prefix)
            out.append(tool)
    return out

tools_from_handlers(app, *, prefixes=None, approval=None)

Build :class:MeshTool entries from the app's @handle registry.

Plumbing under .istos/ is skipped. Pass prefixes to whitelist (exact key expressions). Schemas come from the same path MCP uses.

A handler registered with @handle(approval=…) produces a tool that needs a human; approval=["some/prefix"] marks extra prefixes from the caller's side, for endpoints that did not declare it themselves.

Source code in src/istos/agent/tools.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def tools_from_handlers(
    app: Any,
    *,
    prefixes: Optional[Sequence[str]] = None,
    approval: Optional[Sequence[str]] = None,
) -> List[MeshTool]:
    """Build :class:`MeshTool` entries from the app's ``@handle`` registry.

    Plumbing under ``.istos/`` is skipped. Pass ``prefixes`` to whitelist
    (exact key expressions). Schemas come from the same path MCP uses.

    A handler registered with ``@handle(approval=…)`` produces a tool that needs
    a human; ``approval=["some/prefix"]`` marks extra prefixes from the caller's
    side, for endpoints that did not declare it themselves.
    """
    allow = set(prefixes) if prefixes is not None else None
    needs_human = set(approval or ())
    out: List[MeshTool] = []
    for h in app._handlers:
        if h.prefix.startswith(".istos/"):
            continue
        if allow is not None and h.prefix not in allow:
            continue
        try:
            schemas = get_function_schemas(h.func)
        except Exception:
            schemas = {}
        params = schemas.get("payload_schema") or {"type": "object", "properties": {}}
        out.append(
            MeshTool(
                h.prefix,
                app=app,
                description=(inspect.getdoc(h.func) or "").strip() or tool_name(h.prefix),
                parameters=params,
                approval=getattr(h, "approval", False) or h.prefix in needs_human,
            )
        )
    return out

tools_from_manifest(app, manifest, *, prefixes=None, approval=None)

Build :class:MeshTool entries from one capability manifest.

manifest is what :meth:Istos.export_capabilities returns (and what :meth:Istos.discover_capabilities collects per service). Only handle entries become tools: a mesh tool is a query_once, so streams, channels, and pub/sub entries are not callable this way and are skipped.

app is the local node — it issues the query; the handler itself lives wherever the manifest came from. An entry the owner marked @handle(approval=…) arrives with its flag intact; approval=[…] adds prefixes this caller wants gated regardless.

Source code in src/istos/agent/tools.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def tools_from_manifest(
    app: Any,
    manifest: dict,
    *,
    prefixes: Optional[Sequence[str]] = None,
    approval: Optional[Sequence[str]] = None,
) -> List[MeshTool]:
    """Build :class:`MeshTool` entries from one capability manifest.

    ``manifest`` is what :meth:`Istos.export_capabilities` returns (and what
    :meth:`Istos.discover_capabilities` collects per service). Only ``handle``
    entries become tools: a mesh tool is a ``query_once``, so streams, channels,
    and pub/sub entries are not callable this way and are skipped.

    ``app`` is the *local* node — it issues the query; the handler itself lives
    wherever the manifest came from. An entry the owner marked
    ``@handle(approval=…)`` arrives with its flag intact; ``approval=[…]`` adds
    prefixes this caller wants gated regardless.
    """
    allow = set(prefixes) if prefixes is not None else None
    needs_human = set(approval or ())
    out: List[MeshTool] = []
    for entry in manifest.get("capabilities") or []:
        if not isinstance(entry, dict) or entry.get("kind") != "handle":
            continue
        prefix = entry.get("prefix")
        if not isinstance(prefix, str) or prefix.startswith(".istos/"):
            continue
        if allow is not None and prefix not in allow:
            continue
        out.append(
            MeshTool(
                prefix,
                app=app,
                description=(entry.get("description") or "").strip() or tool_name(prefix),
                parameters=entry.get("params_schema")
                or {"type": "object", "properties": {}},
                approval=entry.get("approval") or prefix in needs_human,
            )
        )
    return out

user_text(msg)

Pull a user string out of a channel message (str or common dict shapes).

Source code in src/istos/agent/loop.py 
62
63
64
65
66
67
68
69
70
71
72
def user_text(msg: Any) -> str:
    """Pull a user string out of a channel message (str or common dict shapes)."""
    if isinstance(msg, str):
        return msg
    if isinstance(msg, dict):
        for key in ("text", "content", "message", "prompt"):
            val = msg.get(key)
            if isinstance(val, str) and val:
                return val
        return json.dumps(msg)
    return str(msg)