Share Resources Across Custom Tools

Custom tools often need one long-lived resource: an HTTP client, a database connection, or a scratch directory. Create it once, give every tool handler the same instance, and register its release with add_cleanup() so the host closes it when the agent stops. Read this page to share a resource across the custom tools of one adapter and to handle the two failure points around it: building the agent and stopping it.

Three example scripts in the SDK repository show the pattern with different resources. Each uses the Python SDK, declares band-sdk>=6.0.1 in its script header, and needs an agent entry in agent_config.yaml and the model provider’s key in the root .env.test.

Prerequisites

• Python 3.11 or later
• The extra for your adapter, for example band-sdk[anthropic], as in the adapter’s page under Connect
• An agent in agent_config.yaml, created as in the Quickstart
• The model provider’s key in .env.test at the repository root. The HTTP client script also calls the public JSONPlaceholder API, so it needs internet access.
• aiosqlite for the SQLite script

Share One Resource Across Tools

A custom tool is an (InputModel, handler) tuple passed as additional_tools. Build the tuples in a function that takes the resource, so every handler closes over the same instance instead of creating its own per call.

import httpx
from pydantic import BaseModel, Field
from band.runtime.custom_tools import CustomToolDef
API_URL = "https://jsonplaceholder.typicode.com"
class GetTodoInput(BaseModel):
    """Read a sample todo from JSONPlaceholder."""
    todo_id: int = Field(ge=1, le=200, description="Todo identifier")
class GetUserInput(BaseModel):
    """Read a sample user's profile from JSONPlaceholder."""
    user_id: int = Field(ge=1, le=10, description="User identifier")
def shared_tools(
    client: httpx.AsyncClient, base_url: str = API_URL
) -> list[CustomToolDef]:
    """Build both handlers around the same live client."""
    base_url = base_url.rstrip("/")
    async def get_todo(args: GetTodoInput) -> str:
        response = await client.get(f"{base_url}/todos/{args.todo_id}")
        response.raise_for_status()
        return response.text
    async def get_user(args: GetUserInput) -> str:
        response = await client.get(f"{base_url}/users/{args.user_id}")
        response.raise_for_status()
        return response.text
    return [(GetTodoInput, get_todo), (GetUserInput, get_user)]

The tools belong to one adapter, so every room that adapter serves shares the resource. The SQLite and temporary-directory scripts say so in their docstrings: all rooms see the same notes or the same draft. Keep room-specific state out of the shared resource.

Register the Cleanup

Create the adapter with the tools, then call add_cleanup() with the callable that releases the resource. In examples/anthropic/shared-resources.py, build_adapter(client, base_url) creates an AnthropicAdapter with additional_tools=shared_tools(client, base_url), calls adapter.add_cleanup(client.aclose) once, and returns the adapter before any host starts. It needs the anthropic extra: pip install "band-sdk[anthropic]".

The callback is a zero-argument async callable. Pass client.aclose, not client.aclose(): passing the result of a call raises TypeError. Register before the host starts, because add_cleanup() raises BandConfigError: Register cleanup before starting the host afterwards. The callback runs when the agent stops, after cleanup_all(), in reverse order of registration. Agent Lifecycle has the full ordering.

Wrap a Synchronous Resource

A callback must return an awaitable. For a resource with a synchronous release, such as TemporaryDirectory, register a small async function that calls it.

In examples/pydantic_ai/draft-workspace.py, build_adapter(workspace) defines async def remove_workspace() that calls workspace.cleanup(), creates a PydanticAIAdapter with additional_tools=shared_tools(Path(workspace.name)), and registers it with adapter.add_cleanup(remove_workspace). That script’s own shared_tools(workspace: Path) builds the draft tools, which write and read a fixed draft.txt inside the workspace, so tool input cannot choose a path outside it. It needs the pydantic-ai extra: pip install "band-sdk[pydantic-ai]".

Release the Resource if Building Fails

Until Agent.from_config() returns, no host owns the resource, so no stop will ever call your callback. If construction raises, release the resource yourself and let the original error propagate. AsyncExitStack does this and hands ownership over cleanly: pop_all() clears the stack once the agent exists.

from contextlib import AsyncExitStack
import aiosqlite
from band import Agent
from band.config.loader import load_agent_config
async def build_agent(settings: Settings) -> Agent:
    """Close construction failures locally; hand off only a complete host."""
    load_agent_config(settings.NOTES_AGENT, config_path=settings.NOTES_CONFIG)
    async with AsyncExitStack() as pending:
        database = await pending.enter_async_context(
            aiosqlite.connect(settings.NOTES_DATABASE, isolation_level=None)
        )
        async with database.execute(NOTES_SCHEMA):
            pass
        agent = Agent.from_config(
            settings.NOTES_AGENT,
            config_path=settings.NOTES_CONFIG,
            adapter=build_adapter(database),
        )
        pending.pop_all()
        return agent

Settings is a pydantic_settings.BaseSettings class that holds the agent name, config path, and database path, and NOTES_SCHEMA is a CREATE TABLE IF NOT EXISTS statement. Both are defined in the script, and build_adapter(database) is that script’s LangGraph adapter builder, which registers database.close (examples/langgraph/shared-database.py). The temporary-directory script uses the same shape with pending.enter_context(workspace). The HTTP client script uses a try/except instead: it closes the client if Agent.from_config() raises, and adds a note to the original error if that close also fails.

Retry an Incomplete Stop

If a cleanup callback raises, the host does not release the resource silently. stop() runs the remaining callbacks, raises an ExceptionGroup whose errors each carry a note naming the failed registration, and keeps the failed callbacks for the next stop(). Retain the agent and call stop() again.

import logging
logger = logging.getLogger(__name__)
async def serve(agent: Agent) -> None:
    """Retry an incomplete stop() once, retaining the host; the run's failure stays chained."""
    try:
        async with agent:
            await agent.run_forever()
    except Exception as original:
        try:
            await agent.stop()
        except Exception:
            message = "Cleanup remains incomplete; retain this Agent and retry stop()"
            original.add_note(message)
            logger.exception("%s", message)
        raise

Until a retry succeeds, starting the same adapter again raises CleanupIncompleteError.

Rebuild After a Stop

After the callbacks have run, the resource is gone. Starting the same adapter again raises BandConfigError telling you to build fresh resources, tool closures, adapter, and host. To run the agent again in the same process, call your build functions from the start: a new client, connection, or directory, then a new adapter and agent.

Two more rules from the SDK:

• An adapter with cleanup callbacks serves one host at a time. Starting a second host on it raises BandConfigError.
• A resource shared across different adapters stays application-owned. Create it in your application, close it yourself, and do not register it with add_cleanup() on each adapter.

Run the Examples

A full run needs live services: Band with the agent entry in agent_config.yaml and a connected room, the model provider, and, for the HTTP client script, the public JSONPlaceholder API. Each script starts with uv run from the repository root, for example:

uv run examples/langgraph/shared-database.py

Then mention the agent in a room. The SQLite script suggests Remember that our release day is Tuesday, then List the team notes, and keeps its notes in notes.sqlite3 in the working directory unless NOTES_DATABASE names another file. The temporary-directory script suggests Save a draft announcing our Tuesday release, then Read the draft, and discards the draft at shutdown. Press Ctrl+C to stop the agent and run the cleanup callbacks.

Next Steps

• Agent Lifecycle: startup order, stop order, and the other adapter hooks
• Create a Framework Adapter: write your own adapter and its hooks