Recipe: Agent with mesh tools

One process exposes tools as @handle. Another (or the same) exposes an agent as @channel. The loop calls tools over Zenoh via query_once.

Tool node

# tools.py
from istos import Istos
from istos.communication.config import IstosZenohConfig

app = Istos(
    service_name="math",
    config=IstosZenohConfig(mode="client", connect_endpoints=["tcp/router:7447"]),
)

@app.handle("math/add")
async def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

@app.handle("math/mul")
async def mul(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b

if __name__ == "__main__":
    app.run()

Agent node

# agent.py
from istos import Istos, ChannelSession, MeshTool
from istos.agent import OpenAIChatModel, drive_channel
from istos.communication.config import IstosZenohConfig

app = Istos(
    service_name="agent",
    http_port=8080,
    config=IstosZenohConfig(mode="client", connect_endpoints=["tcp/router:7447"]),
)

# Tools live on the math node — same prefixes, this app's query_once.
tools = [
    MeshTool(
        "math/add",
        app=app,
        description="Add two integers",
        parameters={
            "type": "object",
            "properties": {
                "a": {"type": "integer"},
                "b": {"type": "integer"},
            },
            "required": ["a", "b"],
        },
    ),
    MeshTool(
        "math/mul",
        app=app,
        description="Multiply two integers",
        parameters={
            "type": "object",
            "properties": {
                "a": {"type": "integer"},
                "b": {"type": "integer"},
            },
            "required": ["a", "b"],
        },
    ),
]

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. Prefer tools over mental arithmetic.",
    )

if __name__ == "__main__":
    app.run()

When the tool handlers are on the same process, use tools_from_handlers(app, prefixes=["math/add", "math/mul"]) instead of hand-built MeshTool entries.

Or let the fabric hand you the catalogue

The schemas above are already published by the math node — capability discovery serves them on .istos/capabilities/<service>. tools_from_discovery reads them and builds the same MeshTool list, so the agent node stops duplicating another service's signatures:

from contextlib import asynccontextmanager
from istos import tools_from_discovery

tools: list = []

@asynccontextmanager
async def on_start(app):
    # Needs an open session, so build the catalogue in the lifespan.
    tools[:] = await tools_from_discovery(app, services=["math"])
    yield

app.lifespan = on_start

@app.channel("agent/chat", ws="/chat", durable=True)
async def chat(s: ChannelSession):
    await drive_channel(s, model, tools, system="…")

Only handle entries become tools (a mesh tool is a query_once; streams and channels are not callable that way). Drop services= for the whole fabric, or pass prefixes= to whitelist exact keys. It is a snapshot — call it again to pick up nodes that joined later, and note that a node started with Istos(enable_discovery=False) does not answer.

Try it

# terminal 1 — zenoh router (or multicast peer mode without a router)
# terminal 2
python tools.py
# terminal 3
python agent.py

# WebSocket client (websocat or similar)
websocat ws://127.0.0.1:8080/chat
> {"text": "what is 6 times 7?"}

Frames on the wire look like {"kind":"tool_call",...}, {"kind":"tool_result",...}, then {"kind":"message","content":"..."}. Pass send_events=False to drive_channel if you only want the final string.

Same pattern with FastAPI in front: Agent channel — bridge the browser socket with open_channel("agent/chat") and leave the loop on the agent node.