Tool calling and 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. Dispatch allows you hook conversation events handlers to workers running on your servers or in the browser.

Tool calling

Tools are code, so they don't go in the agent folder. They run in a worker: a process on your server that connects out to Stream, so nothing you run needs to be publicly reachable. Stream sends each tool call to your worker and the result back to the model. An agent that uses only built-in features and connectors needs no worker.

Install the server SDK

Install the SDK in your server project:

Node.js
npm install @stream-io/stream-agents
Python
uv add "vision-agents[stream]"
Ruby
bundle add getstream-vision-agents
PHP
composer require getstream/vision-agents amphp/websocket-client amphp/http-client-psr7
Go
go get github.com/GetStream/Vision-Agents/sdks/go@latest
.NET
dotnet add package getstream-vision-agents-net
Rust
cargo add vision-agents --git https://github.com/GetStream/Vision-Agents --branch accelerate
cargo add tokio --features full
cargo add serde_json

Set your app's credentials

The worker reads your app's API key and secret from the environment. Set them in the terminal where you run it:

export STREAM_API_KEY=your_api_key
export STREAM_API_SECRET=your_api_secret

The secret is for your server only: never put it in an app or a web page. See Auth.

Write and start the worker

This worker registers a weather_lookup tool on my-agent and waits for calls. Save it in your server project, such as worker.ts:

Node.js
import { Client, Dispatch } from "@stream-io/stream-agents";

const client = new Client();
const agent = client.agent("my-agent");

agent.tools.register<{ location: string }>({
  name: "weather_lookup",
  description: "Look up the current weather for a location",
  parameters: {
    type: "object",
    properties: {
      location: { type: "string", description: "a city or region" },
    },
    required: ["location"],
  },
  run: ({ location }) => {
    const place = location.toLowerCase();
    if (place.includes("colorado")) return "sunny";
    if (place.includes("amsterdam")) return "rainy";
    if (place.includes("san francisco")) return "foggy";
    return `no forecast for ${location}`;
  },
});

await new Dispatch().host(agent).run();
Python
import asyncio

from vision_agents.plugins import stream

agent = stream.Client().agent("my-agent")


@agent.register(description="Look up the current weather for a location")
async def weather_lookup(location: str) -> str:
    place = location.lower()
    if "colorado" in place:
        return "sunny"
    if "amsterdam" in place:
        return "rainy"
    if "san francisco" in place:
        return "foggy"
    return f"no forecast for {location}"


async def main() -> None:
    dispatch = stream.Dispatch()
    dispatch.host(agent)
    await dispatch.run()


asyncio.run(main())
Ruby
require "getstream/vision_agents"

agent = GetStream::VisionAgents::Client.new.agent("my-agent")

agent.tools.register(
  "weather_lookup",
  description: "Look up the current weather for a location",
  parameters: {
    type: "object",
    properties: {
      location: { type: "string", description: "a city or region" },
    },
    required: ["location"],
  },
) do |args|
  location = args["location"]
  place = location.downcase
  if place.include?("colorado") then "sunny"
  elsif place.include?("amsterdam") then "rainy"
  elsif place.include?("san francisco") then "foggy"
  else "no forecast for #{location}"
  end
end

dispatch = GetStream::VisionAgents::Dispatch.new
dispatch.host(agent)
dispatch.run
PHP
<?php

declare(strict_types=1);

use GetStream\VisionAgents\Client;
use GetStream\VisionAgents\Worker\Dispatch;

require __DIR__ . '/vendor/autoload.php';

$agent = (new Client())->agent('my-agent');

$agent->tools->register(
    'weather_lookup',
    'Look up the current weather for a location',
    [
        'type' => 'object',
        'properties' => [
            'location' => [
                'type' => 'string',
                'description' => 'a city or region',
            ],
        ],
        'required' => ['location'],
    ],
    function (array $args): string {
        $location = $args['location'];
        $place = strtolower($location);
        return match (true) {
            str_contains($place, 'colorado') => 'sunny',
            str_contains($place, 'amsterdam') => 'rainy',
            str_contains($place, 'san francisco') => 'foggy',
            default => "no forecast for {$location}",
        };
    },
);

(new Dispatch())->host($agent)->run();
Go
package main

import (
	"context"
	"log"
	"strings"

	"github.com/GetStream/Vision-Agents/sdks/go/agents"
	"github.com/GetStream/Vision-Agents/sdks/go/client"
	"github.com/GetStream/Vision-Agents/sdks/go/stream"
	"github.com/GetStream/Vision-Agents/sdks/go/tools"
)

type WeatherLookup struct {
	Location string `json:"location" schema:"a city or region"`
}

func weatherLookup(_ context.Context, in WeatherLookup) (any, error) {
	place := strings.ToLower(in.Location)
	switch {
	case strings.Contains(place, "colorado"):
		return "sunny", nil
	case strings.Contains(place, "amsterdam"):
		return "rainy", nil
	case strings.Contains(place, "san francisco"):
		return "foggy", nil
	}
	return "no forecast for " + in.Location, nil
}

func main() {
	api, err := client.New(stream.Backend{})
	if err != nil {
		log.Fatal(err)
	}
	agent := api.Agent("my-agent")
	if err := tools.Register(agent.Tools(), "weather_lookup", "Look up the current weather for a location", weatherLookup); err != nil {
		log.Fatal(err)
	}

	dispatch, err := agents.NewDispatch(agents.DispatchOptions{})
	if err != nil {
		log.Fatal(err)
	}
	dispatch.Host(agent, 0)
	log.Fatal(dispatch.Run(context.Background()))
}
.NET
using GetStream.VisionAgents;

using var client = new VisionAgentsClient();
var agent = client.Agent("my-agent");

agent.Tools.Register<WeatherLookup, string>("weather_lookup",
    "Look up the current weather for a location",
    (lookup, _) => Task.FromResult(Forecast(lookup.Location)));

await new Dispatch(new DispatchOptions { Client = client }).Host(agent).RunAsync();

static string Forecast(string location) => location.ToLowerInvariant() switch
{
    var place when place.Contains("colorado") => "sunny",
    var place when place.Contains("amsterdam") => "rainy",
    var place when place.Contains("san francisco") => "foggy",
    _ => $"no forecast for {location}",
};

record WeatherLookup(string Location);
Rust
use serde_json::json;
use vision_agents::{Client, Dispatch};

#[tokio::main]
async fn main() -> vision_agents::Result<()> {
    let client = Client::from_env()?;
    let agent = client.agent("my-agent");

    agent.tools.register(
        "weather_lookup",
        "Look up the current weather for a location",
        json!({
            "type": "object",
            "properties": { "location": { "type": "string", "description": "a city or region" } },
            "required": ["location"],
        }),
        |args| async move {
            let place = args["location"].as_str().unwrap_or_default().to_lowercase();
            let weather = if place.contains("colorado") {
                "sunny"
            } else if place.contains("amsterdam") {
                "rainy"
            } else if place.contains("san francisco") {
                "foggy"
            } else {
                "no forecast for that location"
            };
            Ok::<_, String>(weather)
        },
    );

    let dispatch = Dispatch::new(client);
    dispatch.host(&agent, None);
    dispatch.run().await
}

Run it and leave it running. Every session of my-agent will now call weather_lookup on the connected worker.

Input schema

The input schema of a tool is defined 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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
import { z } from "zod";

agent.tools.register({
  name: "get_weather",
  description: "Current weather in a city",
  parameters: z.object({
    city: z.string().describe("City name, e.g. Amsterdam"),
  }),
  run: async ({ city }) => weather(city),
});
Python
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)
Go
type Weather struct {
	City string `json:"city" schema:"City name, e.g. Amsterdam"`
}

err := tools.Register(agent.Tools(), "get_weather", "Current weather in a city",
	func(ctx context.Context, in Weather) (any, error) {
		return weather(ctx, in.City)
	})
Kotlin
// AgentTool.strings writes the schema of a tool whose arguments are all strings.
// For any other schema, pass a JsonObject. Read each argument defensively.
val getWeather = AgentTool(
    name = "get_weather",
    description = "Current weather in a city",
    parameters = AgentTool.strings(mapOf("city" to "City name, e.g. Amsterdam"), required = listOf("city")),
) { args -> weather(args["city"]?.jsonPrimitive?.content.orEmpty()) }
Swift
// .strings writes the schema of a tool whose arguments are all strings; any other schema is
// a RawJSON literal. Arguments are not checked against it, so read each one defensively.
let getWeather = AgentTool(
    name: "get_weather",
    description: "Current weather in a city",
    parameters: .strings(["city": "City name, e.g. Amsterdam"], required: ["city"])
) { arguments in
    try await weather(arguments["city"]?.stringValue ?? "")
}
Dart
// AgentTool.strings writes the schema of a tool whose arguments are all strings.
// For any other schema, pass a Map. Read each argument defensively.
final getWeather = AgentTool(
  name: 'get_weather',
  description: 'Current weather in a city',
  parameters: AgentTool.strings({'city': 'City name, e.g. Amsterdam'}, required: ['city']),
  run: (args) => weather('${args['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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
import { z } from "zod";

agent.tools.register({
  name: "book_flight",
  description: "Book a flight for the caller",
  parameters: z.object({
    flight_id: z
      .string()
      .regex(/^[A-Z]{2}\d{1,4}$/)
      .describe("e.g. LH400"),
    seats: z.number().int().min(1).max(9),
    cabin: z.enum(["economy", "business", "first"]).default("economy"),
  }),
  run: async ({ flight_id, seats, cabin }) =>
    bookings.book(flight_id, seats, cabin),
});
Python
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)
Go
// Tags: schema (the description), enum, minimum, maximum, exclusiveMinimum,
// exclusiveMaximum, multipleOf, minLength, maxLength, minItems, maxItems,
// pattern, format, default and example. An invopop-style
// `jsonschema:"description=...,minLength=1"` tag works too.
// default is only told to the model: a field left out arrives as its zero value.
type BookFlight struct {
	FlightID string `json:"flight_id" schema:"e.g. LH400" pattern:"^[A-Z]{2}[0-9]{1,4}$"`
	Seats    int    `json:"seats" minimum:"1" maximum:"9"`
	Cabin    string `json:"cabin,omitempty" enum:"economy,business,first" default:"economy"`
}

err := tools.Register(agent.Tools(), "book_flight", "Book a flight for the caller",
	func(ctx context.Context, in BookFlight) (any, error) {
		return airline.Book(ctx, in.FlightID, in.Seats, cmp.Or(in.Cabin, "economy"))
	})

You can also pass the JSON Schema directly:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
// The schema is told to the model as written; arguments are not checked against it,
// so the type parameter is a promise, not a guarantee.
agent.tools.register<{ flight_id: string; seats: number; cabin?: string }>({
  name: "book_flight",
  description: "Book a flight for the caller",
  parameters: {
    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"],
  },
  run: async ({ flight_id, seats, cabin = "economy" }) =>
    bookings.book(flight_id, seats, cabin),
});
Python
@agent.llm.function_registry.register(
    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)
Go
type BookFlightArgs struct {
	FlightID string `json:"flight_id"`
	Seats    int    `json:"seats"`
	Cabin    string `json:"cabin"`
}

schema := []byte(`{
  "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 can be a map[string]any, raw JSON or another library's schema type.
// Arguments are checked against it, then decoded into the struct;
// tools.RegisterSchema hands the function the raw JSON instead.
err := tools.RegisterWithSchema(agent.Tools(), "book_flight", "Book a flight for the caller", schema,
	func(ctx context.Context, in BookFlightArgs) (any, error) {
		return airline.Book(ctx, in.FlightID, in.Seats, cmp.Or(in.Cabin, "economy"))
	})
Kotlin
// The schema is told to the model as written.
val bookFlight = AgentTool(
    name = "book_flight",
    description = "Book a flight for the caller",
    parameters = Json.parseToJsonElement(
        """
        {
          "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"]
        }
        """,
    ).jsonObject,
) { args ->
    bookings.book(
        args["flight_id"]?.jsonPrimitive?.content.orEmpty(),
        args["seats"]?.jsonPrimitive?.intOrNull ?: 1,
        args["cabin"]?.jsonPrimitive?.content ?: "economy",
    )
}
Swift
// The schema is told to the model as written; arguments are not checked against it.
let bookFlight = AgentTool(
    name: "book_flight",
    description: "Book a flight for the caller",
    parameters: [
        "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"],
    ]
) { arguments in
    try await bookings.book(
        arguments["flight_id"]?.stringValue ?? "",
        Int(arguments["seats"]?.numberValue ?? 1),
        arguments["cabin"]?.stringValue ?? "economy"
    )
}
Dart
// The schema is told to the model as written; arguments are not checked against it.
final bookFlight = AgentTool(
  name: 'book_flight',
  description: 'Book a flight for the caller',
  parameters: {
    'type': 'object',
    'properties': {
      'flight_id': {'type': 'string', 'pattern': r'^[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'],
  },
  run: (args) => bookings.book(
    '${args['flight_id'] ?? ''}',
    num.tryParse('${args['seats']}')?.toInt() ?? 1,
    '${args['cabin'] ?? 'economy'}',
  ),
);
Go only

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.

Tools that run in the browser

In JavaScript, agent.tools.register hosts a tool in a Dispatch worker on your server, 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, Swift and Dart tools above run on the device. Only their answers leave the page:

Not documented for Kotlin yet. Showing React.

Not documented for Swift yet. Showing React.

Not documented for Flutter yet. Showing React.

React Native
import { Tools } from "@stream-io/stream-agents";
import * as Location from "expo-location";
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: async () => {
      // The system asks for the permission the first time.
      const { granted } = await Location.requestForegroundPermissionsAsync();
      if (!granted) throw new Error("The person didn't share their location.");
      const { coords } = await Location.getCurrentPositionAsync();
      return { latitude: coords.latitude, longitude: coords.longitude };
    },
  });

// `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 the app restarts, gets them back with its id.
const resumed = await client
  .agent("support")
  .sessions.resume(sessionId, { tools });
React
import { Client, Tools } from "@stream-io/stream-agents";
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: async () => {
      const { coords } = await new Promise<GeolocationPosition>(
        (resolve, reject) =>
          navigator.geolocation.getCurrentPosition(resolve, reject),
      );
      return { latitude: coords.latitude, longitude: coords.longitude };
    },
  });

// `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 });
React Native only

In a React Native app, the same Tools run on the phone. This one reads the location with expo-location, which needs its config plugin and permission texts. The Chat SDK's personal assistant builds an app on such 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 with session.decide, as in Approval.

Concurrency and capacity

A worker has two limits, and they count different work.

Capacity is how many calls and messages a worker's Dispatch handlers work on at once, 4 by default. Each one holds a place until its handler returns. Stream sends new work to the worker using the smallest share of its capacity and passes over a worker that is full, so you can run several workers to share the load. When every worker is full, the call or message fails instead of waiting.

Node.js
const dispatch = new Dispatch({ capacity: 8 });
Python
dispatch = stream.Dispatch(capacity=8)
Ruby
dispatch = GetStream::VisionAgents::Dispatch.new(capacity: 8)
PHP
$dispatch = new Dispatch(capacity: 8);
Go
dispatch, err := agents.NewDispatch(agents.DispatchOptions{Capacity: 8})
.NET
var dispatch = new Dispatch(new DispatchOptions { Capacity = 8 });
Rust
let dispatch = Dispatch::with_capacity(Client::from_env()?, 8);

Concurrency is how many hosted tool calls a worker runs at once. In JavaScript and Go, up to 16 of an agent's calls run at once by default. You can change that for the agent and cap a single tool lower. A call over a cap waits its turn, and its timeout keeps counting while it does. Stream sends each new call to the worker with the fewest still unanswered.

Not in the Python SDK yet. Showing Node.js.

Not in the Ruby SDK yet. Showing Node.js.

Not in the PHP SDK yet. Showing Node.js.

Not in the .NET SDK yet. Showing Node.js.

Not in the Rust SDK yet. Showing Node.js.

Node.js
// At most 32 of this agent's tool calls at once, and 2 bookings.
agent.tools.setConcurrency(32);

agent.tools.register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: BookFlight,
  concurrency: 2,
  run: async ({ flight_id }) => airline.book({ flightId: flight_id }),
});
Go
// At most 32 of this agent's tool calls at once, and 2 bookings.
if err := agent.Tools().SetConcurrency(32); err != nil {
	log.Fatal(err)
}

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

Tool call context

Tool functions are called with the input from the user and a context object with these fields:

  • 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.
Node.js

run is called with the input, then the context. A tool that doesn't need the context leaves it out.

agent.tools.register({
  name: "lookup_order",
  description: "Look up one of the user's orders",
  parameters: LookupOrder,
  run: ({ order_id }, { user }) => orders.find(order_id, { userId: user.id }),
});
Python

Annotate the first parameter with ToolCallContext.

from vision_agents.core.llm import ToolCallContext


@agent.llm.register_function(description="Look up one of the user's orders")
async def lookup_order(tool_call_context: ToolCallContext, order_id: str) -> Order:
    return await orders.find(order_id, user_id=tool_call_context.user.id)
Ruby

A block that takes two arguments is called with the context first, then the input.

agent.tools.register("lookup_order", description: "Look up one of the user's orders",
                     parameters: lookup_order) do |tool_call_context, args|
  Orders.find(args["order_id"], user_id: tool_call_context.user.id)
end
PHP

A callable that takes two arguments, or a ToolCallContext first, is called with the context first, then the input.

use GetStream\VisionAgents\ToolCallContext;

$agent->tools->register('lookup_order', 'Look up one of the user\'s orders', $lookupOrder,
    fn (ToolCallContext $toolCallContext, array $args) => $orders->find($args['order_id'], $toolCallContext->user->id));
Go

Read it from the function's ctx with tools.ToolCallContextFrom.

err = tools.Register(agent.Tools(), "lookup_order", "Look up one of the user's orders",
	func(ctx context.Context, args LookupOrder) (any, error) {
		toolCallContext := tools.ToolCallContextFrom(ctx)
		return orders.Find(ctx, args.OrderID, toolCallContext.User.ID)
	},
)
.NET

Use the overload whose function takes a ToolCallContext first, then the input.

agent.Tools.Register<LookupOrder, Order>("lookup_order", "Look up one of the user's orders",
    (toolCallContext, args, cancellationToken) =>
        orders.FindAsync(args.OrderId, toolCallContext.User.Id, cancellationToken));
Rust

Use register_with_tool_call_context, whose function is called with the context first, then the input.

agent.tools.register_with_tool_call_context(
    types::SessionTool {
        name: "lookup_order".into(),
        description: "Look up one of the user's orders".into(),
        parameters: lookup_order.as_object().cloned().unwrap_or_default(),
        ..Default::default()
    },
    |tool_call_context: ToolCallContext, args| async move {
        orders::find(args["order_id"].as_str().unwrap_or_default(), &tool_call_context.user.id).await
    },
)?;
Kotlin

A tool runs on the user's device, and run is handed the arguments only, as a JsonObject.

Swift

A tool runs on the user's device, and run is handed the arguments only.

Dart

A tool runs on the user's device, and run is handed the arguments only, as a Map<String, Object?>.

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. Each {argument} in the question is filled in from the call:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
agent.tools.register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: BookFlight,
  approval: { title: "Book flight {flight_id}?" },
  run: async ({ flight_id }, toolCallContext) => {
    const booking = await airline.book({
      flightId: flight_id,
      userId: toolCallContext.user.id,
    });
    return `Booked. Confirmation number ${booking.id}.`;
  },
});
Python
from vision_agents.core.llm import ToolCallContext


@agent.llm.register_function(
    description="Book the flight the user picked",
    needs_approval="Book flight {flight_id}?",
)
async def book_flight(tool_call_context: ToolCallContext, flight_id: str) -> str:
    booking = await airline.book(flight_id=flight_id, user_id=tool_call_context.user.id)
    return f"Booked. Confirmation number {booking.id}."
Go
err = tools.Register(agent.Tools(), "book_flight", "Book the flight the user picked",
	func(ctx context.Context, args BookFlight) (any, error) {
		toolCallContext := tools.ToolCallContextFrom(ctx)
		booking, err := airline.Book(ctx, airline.BookRequest{
			FlightID: args.FlightID,
			UserID:   toolCallContext.User.ID,
		})
		if err != nil {
			return nil, err
		}
		return fmt.Sprintf("Booked. Confirmation number %s.", booking.ID), nil
	},
	tools.NeedsApproval("Book flight {flight_id}?"),
)

In a text conversation, the question goes to your app to show. Your app can answer it for the tools it brings itself, like the tools that run in the browser. Each call waits in session.approvals until the user answers it with session.decide:

ReactReact Native
import { Tools } from "@stream-io/stream-agents";
import { z } from "zod";

const tools = new Tools().register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: z.object({ flight_id: z.string().describe("e.g. LH400") }),
  approval: { title: "Book flight {flight_id}?" },
  run: async ({ flight_id }) => {
    const booking = await airline.book(flight_id);
    return `Booked. Confirmation number ${booking.id}.`;
  },
});

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

// In a text conversation, each call waits in session.approvals until the user answers it.
const request = session.approvals[0];
if (request) {
  session.decide(request.id, { allowed: true });
}
Kotlin
val bookFlight = AgentTool(
    name = "book_flight",
    description = "Book the flight the user picked",
    parameters = AgentTool.strings(mapOf("flight_id" to "e.g. LH400"), required = listOf("flight_id")),
    approval = AgentTool.Approval(title = "Book flight {flight_id}?"),
) { args ->
    val booking = airline.book(args["flight_id"]?.jsonPrimitive?.content.orEmpty())
    "Booked. Confirmation number ${booking.id}."
}

// In a text conversation, each call waits in session.approvals until the user answers it.
session.approvals.value.firstOrNull()?.let { request ->
    session.decide(request.id, allowed = true)
}
Swift
let bookFlight = AgentTool(
    name: "book_flight",
    description: "Book the flight the user picked",
    parameters: .strings(["flight_id": "e.g. LH400"], required: ["flight_id"]),
    approval: AgentTool.Approval(title: "Book flight {flight_id}?")
) { arguments in
    let booking = try await airline.book(flightID: arguments["flight_id"]?.stringValue ?? "")
    return "Booked. Confirmation number \(booking.id)."
}

// In a text conversation, each call waits in session.approvals until the user answers it.
if let request = session.approvals.first {
    try await session.decide(request.id, allowed: true)
}
Dart
final bookFlight = AgentTool(
  name: 'book_flight',
  description: 'Book the flight the user picked',
  parameters: AgentTool.strings({'flight_id': 'e.g. LH400'}, required: ['flight_id']),
  approval: const ToolApproval(title: 'Book flight {flight_id}?'),
  run: (args) async {
    final booking = await airline.book('${args['flight_id'] ?? ''}');
    return 'Booked. Confirmation number ${booking.id}.';
  },
);

// `agents` is the StreamAgents the user signed in with.
final session = await agents.agent('support').chat(SessionOptions(tools: [bookFlight]));

// In a text conversation, each call waits in session.approvals until the user answers it.
if (session.approvals.value.firstOrNull case final request?) {
  session.decide(request.id, allowed: true);
}

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.

Warning:

In a text conversation, the answer comes from the process that runs the tool. Only the client SDKs collect it, for the tools the app registered for that session. A tool your server registers, through agent.tools.register or Python's needs_approval, is asked about out loud on voice calls, but no SDK answers its question in a text conversation yet. Its call never reaches session.approvals, and session.decide or an approver's Allow does nothing for it. To ask before such an action in a text conversation, register the tool in your app instead, or have the agent confirm in its reply first.

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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
agent.tools.register({
  name: "search_flights",
  description: "Search for flights",
  parameters: SearchFlights,
  interruptible: true,
  // the call is cancelled when the user talks over the agent
  run: (args) => airline.search(args),
});
Python
@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)
Go
err = tools.Register(agent.Tools(), "search_flights", "Search for flights",
	func(ctx context.Context, args SearchFlights) (any, error) {
		// ctx is cancelled when the user talks over the agent
		return airline.Search(ctx, args)
	},
	tools.Interruptible(true),
)

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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
agent.tools.register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: BookFlight,
  interruptible: true,
  run: async ({ flight_id }, toolCallContext) => {
    const seat = await airline.holdSeat(flight_id);
    // from here on, talking over the agent doesn't cancel the booking
    await toolCallContext.disallowInterruptions();
    return airline.book({ flightId: flight_id, seat });
  },
});
Python
from vision_agents.core.llm import ToolCallContext


@agent.llm.register_function(description="Book the flight the user picked", interruptible=True)
async def book_flight(tool_call_context: ToolCallContext, flight_id: str) -> str:
    seat = await airline.hold_seat(flight_id)
    # from here on, talking over the agent doesn't cancel the booking
    await tool_call_context.disallow_interruptions()
    booking = await airline.book(flight_id=flight_id, seat=seat)
    return f"Booked. Confirmation number {booking.id}."
Go
err = tools.Register(agent.Tools(), "book_flight", "Book the flight the user picked",
	func(ctx context.Context, args BookFlight) (any, error) {
		seat, err := airline.HoldSeat(ctx, args.FlightID)
		if err != nil {
			return nil, err
		}
		// from here on, talking over the agent doesn't cancel ctx
		ctx = tools.ToolCallContextFrom(ctx).DisallowInterruptions(ctx)
		return airline.Book(ctx, airline.BookRequest{FlightID: args.FlightID, Seat: seat})
	},
	tools.Interruptible(true),
)

When the call was cancelled before it could be held, the change never starts: JavaScript rejects with CallNotHeldError, 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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
agent.tools.register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: BookFlight,
  timeoutMs: 60_000,
  run: async ({ flight_id }) => airline.book({ flightId: flight_id }),
});
Python
@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)
Go
err = tools.Register(agent.Tools(), "book_flight", "Book the flight the user picked",
	func(ctx context.Context, args BookFlight) (any, error) {
		return airline.Book(ctx, airline.BookRequest{FlightID: args.FlightID})
	},
	tools.Timeout(60*time.Second),
)

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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
agent.tools.register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: BookFlight,
  run: async ({ flight_id }, toolCallContext) => {
    toolCallContext.update("Checking seats on the flight");
    const booking = await airline.book({ flightId: flight_id });
    return `Booked. Confirmation number ${booking.id}.`;
  },
});
Python
from vision_agents.core.llm import ToolCallContext


@agent.llm.register_function(description="Book the flight the user picked")
async def book_flight(tool_call_context: ToolCallContext, flight_id: str) -> str:
    await tool_call_context.update("Checking seats on the flight")
    booking = await airline.book(flight_id=flight_id)
    return f"Booked. Confirmation number {booking.id}."
Go
err = tools.Register(agent.Tools(), "book_flight", "Book the flight the user picked",
	func(ctx context.Context, args BookFlight) (any, error) {
		tools.ToolCallContextFrom(ctx).Update(ctx, "Checking seats on the flight")
		booking, err := airline.Book(ctx, airline.BookRequest{FlightID: args.FlightID})
		if err != nil {
			return nil, err
		}
		return fmt.Sprintf("Booked. Confirmation number %s.", booking.ID), nil
	},
)

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:

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
agent.tools.register({
  name: "book_flight",
  description: "Book the flight the user picked",
  parameters: BookFlight,
  run: async ({ flight_id }, toolCallContext) => {
    const booking = await airline.book({
      flightId: flight_id,
      userId: toolCallContext.user.id,
      idempotencyKey: toolCallContext.id,
    });
    return `Booked. Confirmation number ${booking.id}.`;
  },
});
Python
from vision_agents.core.llm import ToolCallContext


@agent.llm.register_function(description="Book the flight the user picked")
async def book_flight(tool_call_context: ToolCallContext, flight_id: str) -> str:
    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}."
Go
err = tools.Register(agent.Tools(), "book_flight", "Book the flight the user picked",
	func(ctx context.Context, args BookFlight) (any, error) {
		toolCallContext := tools.ToolCallContextFrom(ctx)
		booking, err := airline.Book(ctx, airline.BookRequest{
			FlightID:       args.FlightID,
			UserID:         toolCallContext.User.ID,
			IdempotencyKey: toolCallContext.ID,
		})
		if err != nil {
			return nil, err
		}
		return fmt.Sprintf("Booked. Confirmation number %s.", booking.ID), nil
	},
)

Returning images

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

Not documented for Ruby yet. Showing Node.js.

Not documented for PHP yet. Showing Node.js.

Not documented for .NET yet. Showing Node.js.

Not documented for Rust yet. Showing Node.js.

Node.js
import { ToolOutput } from "@stream-io/stream-agents";

agent.tools.register({
  name: "seat_map",
  description: "Show the seat map for a flight",
  parameters: SeatMap,
  run: async ({ flight_id }) =>
    new ToolOutput({
      text: `Seat map for ${flight_id}`,
      images: [{ url: await airline.seatMapUrl(flight_id) }],
    }),
});
Python
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")]
Go
err = tools.Register(agent.Tools(), "seat_map", "Show the seat map for a flight",
	func(ctx context.Context, args SeatMap) (any, error) {
		url, err := airline.SeatMapURL(ctx, args.FlightID)
		if err != nil {
			return nil, err
		}
		return tools.Output{Text: "Seat map for " + args.FlightID, Images: []tools.Image{{URL: url}}}, nil
	},
)

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

Dispatch workers can register an event handler function and run it on your side. 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:

Node.js
import { Agent, Dispatch } from "@stream-io/stream-agents";

const dispatch = new Dispatch({ capacity: 8 });
const support = () =>
  new Agent({ name: "support", pipeline: { config: "support" } });

// Answer each phone call with a support agent until the caller hangs up.
dispatch.onCall(async (call) => {
  const session = await support().answer(call);
  await session.wait();
});

// Handle each message written to the agent.
dispatch.onMessage(async (message) => {
  // A session is already running: have the model reply on it.
  if (message.sessionId) {
    await dispatch.answer(message);
    return;
  }
  // Get the session answering the message's channel, started if none is.
  const session = await dispatch.sessionFor(message, support);
  await session.responses.create(message.text);
});

// Run the agent's tools for every session of it, wherever it was opened.
dispatch.host(agent);

await dispatch.run();
Python
from vision_agents.core import Agent
from vision_agents.plugins import stream

dispatch = stream.Dispatch(capacity=8)


# Answer each phone call with a support agent until the caller hangs up.
@dispatch.wait_for_call()
async def rang(call: stream.InboundCall) -> None:
    agent = Agent(config="support")
    async with agent.answer(call):
        await agent.responses.create("Greet the caller")


# Handle each message written to the agent.
@dispatch.wait_for_message()
async def written(message: stream.InboundMessage) -> None:
    # A session is already running: have the model reply on it.
    if message.session_id:
        await dispatch.answer(message)
        return
    # Get the session answering the message's channel, started if none is.
    agent = await dispatch.get_or_create_agent(message, lambda: Agent(config="support"))
    await agent.responses.create(message.text)


# Run the agent's tools for every session of it, wherever it was opened.
dispatch.host(agent)

await dispatch.run()
Ruby
dispatch = GetStream::VisionAgents::Dispatch.new(capacity: 8)

# Answer each phone call with a support agent until the caller hangs up.
dispatch.wait_for_call do |call|
  GetStream::VisionAgents::Agent.new(config: "support").join(call)
end

# Handle each message written to the agent.
dispatch.wait_for_message do |message|
  # A session is already running: have the model reply on it.
  next dispatch.answer(message) unless message.session_id.empty?

  # Get the session answering the message's channel, started if none is.
  dispatch.get_or_create_agent(message) { GetStream::VisionAgents::Agent.new(config: "support") }
          .reply(message)
          .responses.create(message.text)
end

# Run the agent's tools for every session of it, wherever it was opened.
dispatch.host(agent)

dispatch.run
PHP
use GetStream\VisionAgents\Agent;
use GetStream\VisionAgents\Inbound\InboundCall;
use GetStream\VisionAgents\Inbound\InboundMessage;
use GetStream\VisionAgents\Worker\Dispatch;

$dispatch = new Dispatch(capacity: 8);
$support = fn (): Agent => new Agent(config: 'support');

// Answer each phone call with a support agent until the caller hangs up.
$dispatch->waitForCall(function (InboundCall $call) use ($support): void {
    $support()->answer($call)->watch()->wait();
});

// Handle each message written to the agent.
$dispatch->waitForMessage(function (InboundMessage $message) use ($dispatch, $support): void {
    // A session is already running: have the model reply on it.
    if ($message->sessionId !== '') {
        $dispatch->answer($message);
        return;
    }
    // Get the session answering the message's channel, started if none is.
    $dispatch->getOrCreateAgent($message, $support)->responses->create($message->text);
});

// Run the agent's tools for every session of it, wherever it was opened.
$dispatch->host($agent);

$dispatch->run();
Go
dispatch, err := agents.NewDispatch(agents.DispatchOptions{Capacity: 8})
if err != nil {
	return err
}

support, err := agents.New(agents.Options{
	Name: "support",
	LLM:  stream.Accelerated(stream.Config{Agent: "support"}),
})
if err != nil {
	return err
}

// Answer each phone call with the support agent until the caller hangs up.
dispatch.OnCall(func(ctx context.Context, call agents.InboundCall) error {
	session, err := support.Answer(ctx, call)
	if err != nil {
		return err
	}
	return session.Wait(ctx)
})

// Handle each message written to the agent.
dispatch.OnMessage(func(ctx context.Context, message agents.InboundMessage) error {
	// A session is already running: have the model reply on it.
	if message.SessionID != "" {
		return dispatch.Answer(ctx, message)
	}
	// Get the conversation answering the message's channel, started if none is.
	conversation, err := dispatch.Conversation(ctx, message,
		func(context.Context, agents.InboundMessage) (*agents.Agent, error) { return support, nil })
	if err != nil {
		return err
	}
	return conversation.Respond(message.Text)
})

// Run the agent's tools for every session of it, wherever it was opened.
dispatch.Host(agent, 0)

return dispatch.Run(ctx)
.NET
var dispatch = new Dispatch(new DispatchOptions { Capacity = 8 });

// Answer each phone call with a support agent until the caller hangs up.
dispatch.WaitForCall(async call =>
{
    await using var session = await new Agent(new AgentOptions { Config = "support" }).JoinAsync(call);
    await session.WaitAsync();
});

// Handle each message written to the agent.
dispatch.WaitForMessage(async message =>
{
    // A session is already running: have the model reply on it.
    if (message.SessionId != "")
    {
        await dispatch.AnswerAsync(message);
        return;
    }
    // Get the session answering the message's channel, started if none is.
    var agent = await dispatch.GetOrCreateAgentAsync(message,
        () => new Agent(new AgentOptions { Config = "support" }));
    await agent.Responses.CreateAsync(message.Text);
});

// Run the agent's tools for every session of it, wherever it was opened.
dispatch.Host(agent);

await dispatch.RunAsync(cancellationToken);
Rust
use vision_agents::{Agent, Client, Dispatch, InboundCall, InboundMessage};

let dispatch = Dispatch::with_capacity(Client::from_env()?, 8);

// Answer each phone call with a support agent until the caller hangs up.
dispatch.wait_for_call(|call: InboundCall| async move {
    Agent::new("support").answer(&call).await?.wait().await;
    Ok(())
});

// Handle each message written to the agent.
let worker = dispatch.clone();
dispatch.wait_for_message(move |message: InboundMessage| {
    let dispatch = worker.clone();
    async move {
        // A session is already running: have the model reply on it.
        if !message.session_id.is_empty() {
            dispatch.answer(&message).await?;
            return Ok(());
        }
        // Get the session answering the message's channel, started if none is.
        let session = dispatch
            .get_or_create_agent(&message, || async { Ok(Agent::new("support")) })
            .await?;
        session.responses.create(&message.text).await?;
        Ok(())
    }
});

// Run the agent's tools for every session of it, wherever it was opened.
dispatch.host(&agent, None);

dispatch.run().await

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

The method names differ a bit per SDK:

What it does Node.js 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

Handling text to your server

By default, the agent harness uses LLMs to answer users message. You can also have your server handle messages first, in agent.yaml:

name: support
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.

Warning:

text: enabled 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.