Agent Lifecycle

An agent built on the Band SDK starts in a fixed order, recovers messages across reconnects and restarts, and stops in a way that lets in-progress work finish. Read this page to run an agent under a process manager or in a container, to place setup and cleanup code in the right hook, or to diagnose an agent that fails to start. Method signatures are in the SDK Reference.

This page describes the Python SDK, band-sdk 6.0.1.

Prerequisites

• Python 3.11 or later
• band-sdk[langgraph]==6.0.1, installed as in LangGraph
• agent_config.yaml with the agent’s ID and API key under my_agent, created as in the Quickstart
• OPENAI_API_KEY in .env

Run an Agent

agent.run() starts the agent, blocks until the process is interrupted, and then stops the agent.

import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from band import Agent, configure_logging
from band.adapters import LangGraphAdapter
async def main() -> None:
    load_dotenv()
    configure_logging()
    adapter = LangGraphAdapter(
        llm=ChatOpenAI(model="gpt-5.4-mini"),
        checkpointer=InMemorySaver(),
    )
    agent = Agent.from_config("my_agent", adapter=adapter)
    await agent.run()
asyncio.run(main())

python agent.py

When the agent is connected and has joined its rooms, the log shows band.agent: Agent started: <agent name> (band-sdk 6.0.1). Press Ctrl+C and the log shows band.agent: Agent stopped: <agent name> (graceful=True).

agent.run() is the same as async with agent: await agent.run_forever(). Use the context manager form when you need your own try/finally inside it, for example to close a client before stop() runs on exit.

Startup Sequence

agent.start() runs these steps in order. agent.run() and async with agent: call it for you.

1. Take the single-instance lock. A second process for the same agent ID on the same host fails here, before the adapter starts anything. See One Process per Agent.
2. Fetch the agent’s metadata. GET /api/v1/agent/me returns the agent’s name, description, and the deployment’s feature flags. A 429 response is retried after 1, 2, 4, 8, and 16 seconds. Capabilities the adapter enabled but the deployment does not serve are dropped.
3. Call on_started(agent_name, agent_description). This runs before the WebSocket opens, so a system prompt built here is ready before the first message arrives.
4. Open the WebSocket. The SDK joins the agent_control:{agent_id} and agent_rooms:{agent_id} channels.
5. Join existing rooms. GET /api/v1/agent/chats lists the rooms the agent is in. For each room the SDK joins chat_room:{room_id} and room_participants:{room_id} and starts that room’s execution context.
6. Catch up on each room. Each room first recovers messages left in processing by an earlier run, then works through the backlog from GET /api/v1/agent/chats/{chat_id}/messages/next until it returns 204.
7. Subscribe to contact events when ContactEventConfig asks for them.

If on_started() or a step after it fails, the SDK stops what it started, runs the adapter’s cleanup_all() and any callbacks registered with add_cleanup(), and releases the lock before the error propagates, so a retry in the same process starts clean. When that cleanup itself fails, the lock stays held, the error carries the note Agent cleanup is incomplete; retain this host and call stop() to retry, and a new start() raises CleanupIncompleteError until stop() succeeds.

Adapter Hooks

Register add_cleanup() callbacks in the adapter’s constructor or before the agent starts. After start() it raises BandConfigError: Register cleanup before starting the host. A callback that fails is retried by the next stop(). After the callbacks have run, the adapter cannot be started again: build fresh resources, adapter, and agent. An adapter with callbacks serves one agent at a time. Create a Framework Adapter shows each hook in a working adapter.

Reconnects and Recovery

Band delivers each message at least once. The same message can reach the agent again after a crash or a reconnect, so make side effects idempotent or deduplicate on the message ID.

Stop an Agent

agent.stop(timeout) waits up to timeout seconds for the message each room is handling, then cancels whatever is still running. It returns True when every room finished in time and False when it had to cancel. With timeout=None it cancels at once.

Stopping runs in this order: each room’s execution context stops and on_cleanup(room_id) runs for it, the SDK leaves the room channels and the contacts channel, the WebSocket closes, cleanup_all() runs, and then the add_cleanup() callbacks run. The lock is released only after every one of these succeeds. If room cleanup fails, stop() still runs cleanup_all(), holds back the add_cleanup() callbacks until room work has settled, raises an ExceptionGroup, and keeps the lock. The next stop() retries only what failed.

agent.run() stops the agent when the process receives SIGINT or SIGTERM, which is what Ctrl+C, docker stop, and Kubernetes send, and passes its shutdown_timeout argument (default 30 seconds) to stop(). The context manager form uses the same 30-second default. These signal handlers are installed only on Unix, in the main thread.

A message that is cancelled during shutdown stays in processing, and the next start handles it again.

Control the Lifecycle Yourself

Call start() and stop() directly when the agent runs inside a larger program or for a fixed time.

import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from band import Agent, configure_logging
from band.adapters import LangGraphAdapter
async def main() -> None:
    load_dotenv()
    configure_logging()
    adapter = LangGraphAdapter(
        llm=ChatOpenAI(model="gpt-5.4-mini"),
        checkpointer=InMemorySaver(),
    )
    agent = Agent.from_config("my_agent", adapter=adapter)
    await agent.start()
    try:
        await asyncio.sleep(300)
    finally:
        graceful = await agent.stop(timeout=30.0)
        print(f"Stopped, graceful={graceful}")
asyncio.run(main())

The agent handles messages for five minutes, then prints Stopped, graceful=True. Without run_forever() no signal handlers are installed. To stop on a signal in this form, create a GracefulShutdown(agent, timeout=30.0) from band, call its register_signals() before start(), and await its wait_for_shutdown() instead of the sleep.

One Process per Agent

Run one process per agent. Two processes with the same agent ID claim each other’s in-flight messages, and adapters that keep sessions on disk resume the same session twice.

The SDK enforces this on one host with a lock file in the process’s temp directory. A second process fails at startup with AgentAlreadyRunningError: Agent <agent_id> is already running on this host (lock: <path>). The lock is released when the process exits, including on a crash. Processes with different temp directories, in different containers, or on different hosts do not see each other’s locks. Agent.create(config=AgentConfig(single_instance=False)) turns the check off.

Band itself keeps one WebSocket connection per agent. When a second connection opens with the same agent’s credentials, Band replaces the first one, and the SDK in that first process logs WebSocket connection superseded and does not reconnect. To have Band refuse the second connection instead, set AgentConfig(conflict_policy=ConflictPolicy.REJECT): the second process fails at startup with AgentAlreadyRunningError: Agent <agent_id> already has a live connection to the platform. One Connection per Agent has the rules.

Startup Errors

For connection and delivery problems after startup, see Troubleshooting.

Next Steps