Tool calling & Dispatch

Tool calling makes it easy to connect your agent to your own systems. A support agent looks up the customer and checks their order status. A legal AI searches your case database.

Tool calling

Register a function in your code and the model can call it. Your process runs the function and sends the result back, so it can reach your database and internal APIs:

@agent.llm.register_function(description="Look up an order by its number")
async def lookup_order(order_id: str) -> Order:
    return await orders.find(order_id)

The arguments are described to the model by a type: a Zod schema in JavaScript, type hints or a Pydantic model in Python, and a struct or class in most other SDKs. You can also pass a JSON Schema directly. The return value is sent back as the tool result. Errors are sent to the model too, so it can tell the user what went wrong.

Services like Slack, Sentry or Google Calendar don't need any code. See the catalog.

Tools that run in the browser

In JavaScript, agent.tools.register hosts a tool in a Dispatch worker on your backend, for every session of the agent. A page can also bring tools of its own, which run in the browser for the session it opens, like the Kotlin and Swift tools above run on the device. Only their answers leave the page:

import { Client, Tools } from "@stream-io/calypso";
import { z } from "zod";

const tools = new Tools()
  .register({
    name: "get_local_context",
    description: "The person's current date, local time and timezone.",
    parameters: z.object({}),
    run: () => ({
      now: new Date().toISOString(),
      timezone: Intl.DateTimeFormat().resolvedOptions().timeZone,
    }),
  })
  .register({
    name: "get_location",
    description:
      "Where the person is. Only call it when the answer depends on it.",
    parameters: z.object({
      reason: z.string().describe("why you need it, in a few words"),
    }),
    displayTitle: "Checking where you are",
    approval: {
      title: "Share your location?",
      reasonArgument: "reason",
      allowTitle: "Share",
      declineTitle: "Not now",
    },
    run: () =>
      new Promise((resolve, reject) =>
        navigator.geolocation.getCurrentPosition(
          ({ coords }) =>
            resolve({ latitude: coords.latitude, longitude: coords.longitude }),
          reject,
        ),
      ),
  });

// `client` holds your apiKey and a user from setUser.
const session = await client.agent("support").sessions.create({ tools });

// A session opened before, such as after a reload, gets them back with its id.
const resumed = await client
  .agent("support")
  .sessions.resume(sessionId, { tools });

The session's socket runs them, so they answer while the page holds the session open. A tool with an approval waits in session.approvals until the user answers it, as in Approval.

Input schema

The input schema of a tool is always a JSON Schema object. Depending on the SDK, you can describe it the idiomatic way instead: Zod in JavaScript, Pydantic in Python and struct tags in Go.

The arguments the model sends are checked against the schema before your function runs. Arguments that don't fit fail the call with what was wrong, such as seats: Input should be less than or equal to 9, and the model can ask again. Your function never sees them.

For simple tools, annotate the arguments and the SDK generates the JSON Schema:

from typing import Annotated

from pydantic import Field


@agent.llm.register_function(description="Current weather in a city")
async def get_weather(
    city: Annotated[str, Field(description="City name, e.g. Amsterdam")],
) -> str:
    return await weather(city)

For a tool with several fields and validation rules, describe them with Zod, Pydantic or Go struct tags. This is the preferred approach for anything that needs more than one field:

from typing import Literal

from pydantic import BaseModel, Field


class BookFlight(BaseModel):
    flight_id: str = Field(pattern=r"^[A-Z]{2}\d{1,4}$", description="e.g. LH400")
    seats: int = Field(ge=1, le=9)
    cabin: Literal["economy", "business", "first"] = "economy"


# A function that takes a single model gets the model's fields as the tool's
# arguments, and is handed the parsed model.
@agent.llm.register_function(description="Book a flight for the caller")
async def book_flight(booking: BookFlight) -> str:
    return await bookings.book(booking.flight_id, booking.seats, booking.cabin)

You can also pass the JSON Schema directly:

@agent.llm.register_function(
    description="Book a flight for the caller",
    parameters_schema={
        "type": "object",
        "properties": {
            "flight_id": {"type": "string", "pattern": "^[A-Z]{2}[0-9]{1,4}$", "description": "e.g. LH400"},
            "seats": {"type": "integer", "minimum": 1, "maximum": 9},
            "cabin": {"type": "string", "enum": ["economy", "business", "first"], "default": "economy"},
        },
        "required": ["flight_id", "seats"],
    },
)
# The schema is told to the model as written; arguments are not checked against it.
async def book_flight(flight_id: str, seats: int, cabin: str = "economy") -> str:
    return await bookings.book(flight_id, seats, cabin)

In Go, a tool can also be a type. Its fields are the arguments, and Run is called with them filled in:

type GetWeather struct {
	City string `json:"city" schema:"City name, e.g. Amsterdam"`
}

func (GetWeather) Name() string        { return "get_weather" }
func (GetWeather) Description() string { return "Current weather in a city" }
func (w GetWeather) Run(ctx context.Context) (any, error) {
	return weather(ctx, w.City)
}

err := agent.Tools().Add(GetWeather{})

Unexported fields are yours, for whatever Run needs to reach, such as a database handle.

Tool call context

Tool functions are called with the input and the context of the call:

  • id: the unique identifier of this call. It stays the same when the call is sent again, so use it to make tool calls idempotent and handle retries.
  • session_id: the session the call comes from.
  • user: the user whose turn triggered the call.
  • conversation: the chat conversation.

It is get_tool_call_context().

Approval

By default, a tool runs as soon as the model calls it. A tool can instead ask the user first. On a voice call the agent asks out loud and runs the tool only on a yes. In a text conversation, the question goes to your app to show. Each {argument} in the question is filled in from the call:

from vision_agents.core.llm import get_tool_call_context


@agent.llm.register_function(
    description="Book the flight the user picked",
    needs_approval="Book flight {flight_id}?",
)
async def book_flight(flight_id: str) -> str:
    tool_call_context = get_tool_call_context()
    booking = await airline.book(flight_id=flight_id, user_id=tool_call_context.user.id)
    return f"Booked. Confirmation number {booking.id}."

In a conversation kept in Stream Chat, the call's step in the reply carries the question too, and the AI components ask it there: pass an approver to AIConversation in React, as the React assistant sample does. session.conversation tells a page when the calls waiting change, including one the model withdrew.

Interruptions

By default, a tool keeps running when the user talks over the agent, and its result reaches the model when it comes. A tool that only reads, such as a search, can ask to be cancelled instead:

@agent.llm.register_function(description="Search for flights", interruptible=True)
async def search_flights(destination: str) -> list[Flight]:
    # the task is cancelled when the user talks over the agent
    return await airline.search(destination)

A tool that changes something can stay interruptible until the change starts. From then on it holds the call, so talking over the agent no longer cancels it and only its timeout does:

from vision_agents.core.llm import get_tool_call_context


@agent.llm.register_function(description="Book the flight the user picked", interruptible=True)
async def book_flight(flight_id: str) -> str:
    seat = await airline.hold_seat(flight_id)
    # from here on, talking over the agent doesn't cancel the booking
    await get_tool_call_context().disallow_interruptions()
    booking = await airline.book(flight_id=flight_id, seat=seat)
    return f"Booked. Confirmation number {booking.id}."

When the call was cancelled before it could be held, the change never starts: in JavaScript the signal is already aborted, Python raises CallNotHeld, and in Go the context comes back cancelled.

Timeouts

By default, a tool call fails and is cancelled when it doesn't return within 30 seconds. You can set a timeout per tool:

@agent.llm.register_function(description="Book the flight the user picked", timeout=60)
async def book_flight(flight_id: str) -> str:
    return await airline.book(flight_id=flight_id)

Concurrency

In Go, up to 16 calls of a session's tools run at once. You can cap a single tool lower. A call over the cap waits its turn, and its timeout keeps counting while it does:

err = tools.Register(agent.Tools(), "book_flight", "Book the flight the user picked", bookFlight,
	tools.Concurrency(2),
)

Progress updates

A slow tool can tell the user how it's going. On a voice call the agent says it in its own words once nobody is talking. In a text conversation it shows on the tool call:

from vision_agents.core.llm import get_tool_call_context


@agent.llm.register_function(description="Book the flight the user picked")
async def book_flight(flight_id: str) -> str:
    await get_tool_call_context().update("Checking seats on the flight")
    booking = await airline.book(flight_id=flight_id)
    return f"Booked. Confirmation number {booking.id}."

A newer update replaces one that hasn't been said yet, and the result replaces both. To have the agent say your exact words instead, use say (Say in Go). A speech-to-speech model may still paraphrase them.

Idempotency

A call can be sent to your process again, for example when it reconnects. The call's id stays the same, so use it as the idempotency key for anything the tool changes:

from vision_agents.core.llm import get_tool_call_context


@agent.llm.register_function(description="Book the flight the user picked")
async def book_flight(flight_id: str) -> str:
    tool_call_context = get_tool_call_context()
    booking = await airline.book(
        flight_id=flight_id,
        user_id=tool_call_context.user.id,
        idempotency_key=tool_call_context.id,
    )
    return f"Booked. Confirmation number {booking.id}."

Returning images

A tool can show the model pictures, such as a seat map. Return them with the text:

from vision_agents.core.llm import ImageContent


@agent.llm.register_function(description="Show the seat map for a flight")
async def seat_map(flight_id: str) -> list[str | ImageContent]:
    png = await airline.seat_map(flight_id)
    return [f"Seat map for {flight_id}", ImageContent(data=png, mime="image/png")]

An image is an https URL or the image itself (bytes in Python, a data URI in JavaScript and Go), at most 5 MB. On a voice call the agent's vision skill looks at it. In a text conversation a model that reads images sees it, and one that reads only text is told the tool failed.

Dispatch

A dispatch worker is a process on your server that receives work from Stream: messages, phone calls and tool calls. It connects out to Stream, so nothing you run needs to be publicly reachable.

Running a worker

Register a handler and run the worker. When a message has a session id, answer lets the model reply on that session. Otherwise sessionFor gives you the session for that channel, and starts one from your agent if needed:

from vision_agents.core import Agent
from vision_agents.plugins import stream

dispatch = stream.Dispatch(capacity=8)


@dispatch.wait_for_message()
async def written(message: stream.InboundMessage) -> None:
    if message.session_id:
        await dispatch.answer(message)
        return
    agent = await dispatch.get_or_create_agent(message, lambda: Agent(config="support"))
    await agent.responses.create(message.text)


await dispatch.run()

To reply in your own words instead of the model's, skip answer and send the reply from your backend.

The method names differ a bit per SDK:

What it does JavaScript Python, Ruby, Rust PHP Go C#
Handle each message onMessage wait_for_message waitForMessage OnMessage WaitForMessage
Handle each incoming call onCall wait_for_call waitForCall OnCall WaitForCall
The session answering a message's channel sessionFor get_or_create_agent getOrCreateAgent Conversation GetOrCreateAgentAsync
Have the model reply on a session answer answer answer Answer AnswerAsync
Host an agent's tools host host host Host Host

Hosting tools

Tools registered on an agent only run for the sessions that process opens. Sessions your users open from a browser or phone need a worker to host their tools. Every session of the agent can then use them:

agent = stream.Client().agent("support")


@agent.register(description="Look up an order by its number")
async def lookup_order(order_id: str) -> Order:
    return await orders.find(order_id)


dispatch.host(agent, tool_timeout=60)

Host tools before calling run. The worker keeps running until you stop it. The tool timeout is how long Stream waits for one tool call to be answered before telling the model it failed: two minutes unless you set it.

Leaving text to your server

By default the model answers every message. You can also have your server handle messages first, in agent.yaml:

name: Stream product
mode: text
dispatch:
  incoming_call: enabled
  text: enabled

With text: enabled each message goes to your worker first. Check it, enrich it or route it, then call answer to let the model reply. The user sees the agent thinking in the meantime.

This adds a round trip before the model starts answering. Messages with images or video are refused, and a message is dropped when no worker is connected. Phone calls to your numbers always go to a worker, so incoming_call doesn't change anything yet.

Capacity

Capacity is how many messages and calls a worker handles at once. Stream sends new work to the worker with the most room, so you can run several workers to share the load. Hosted tool calls don't count against capacity.