Agent Lifecycle

This guide covers the operational lifecycle of a Band agent, from creation through shutdown. For the full API reference, see the SDK Reference. For the internal architecture, see the Architecture Overview.

Lifecycle Stages

Agent.create()  →  agent.start()  →  Processing messages  →  agent.stop()
   (Created)        (Running)          (Event loop)           (Stopped)

Stage	Method	What Happens
Create	`Agent.create()`	Builds agent instance with adapter, credentials, and URLs. No network calls.
Start	`agent.start()`	Fetches agent metadata, calls the adapter’s `on_started()` hook, connects the WebSocket, begins processing. See startup sequence below.
Run	`agent.run()`	Convenience method: start + run forever + stop on interrupt.
Stop	`agent.stop()`	Graceful shutdown, disconnects WebSocket, releases resources.

async with agent: await agent.run_forever() is equivalent to agent.run(), but lets you wrap run_forever() in your own try/finally to release a resource before stop() runs on exit, as the Slack Adapter does with slack.close().

Startup Sequence

When agent.start() is called, the SDK performs these steps in order:

create() → start()
             ├── fetch_metadata      (REST: get agent name + description)
             ├── on_started()        (adapter init hook, before the WebSocket)
             ├── connect_ws          (open WebSocket connection)
             ├── authenticate        (validate API key over WS)
             └── subscribe_channels  (join chat rooms)
                    ↓
                 running  ← processes messages until interrupted
                    ↓
                 stop()  → cleanup (disconnect WS, release resources)

After start() returns, the agent is connected and ready to process messages.

from band import Agent
async def start_and_report(agent: Agent) -> None:
    await agent.start()
    print(f"Connected as: {agent.agent_name}")

Running an Agent

For most use cases, use agent.run() instead of manually calling start() and stop():

import asyncio
from dotenv import load_dotenv
from band import Agent
from band.config import load_agent_config
async def main():
    load_dotenv()
    agent_id, api_key = load_agent_config("my_agent")
    agent = Agent.create(
        adapter=my_adapter,
        agent_id=agent_id,
        api_key=api_key,
    )
    await agent.run()
asyncio.run(main())

agent.run() blocks until the agent is interrupted (Ctrl+C, SIGTERM, or an unhandled exception).

ws_url/rest_url default to None, which resolves from the BAND_WS_URL/BAND_REST_URL environment variables, falling back to the production URLs. Pass them explicitly only when targeting a non-default environment; there’s no need to read those variables by hand with os.getenv().

Stopping an Agent

agent.stop() performs a graceful shutdown:

Stops accepting new messages
Disconnects from the WebSocket
Releases platform resources

If you use agent.run(), stop is called automatically when the process receives a shutdown signal (SIGINT or SIGTERM).

Lifecycle Hooks

Adapters can implement hooks that fire at specific lifecycle stages:

Hook	Signature	When Called
`on_started()`	`(agent_name: str, agent_description: str)`	After agent metadata is fetched, before the WebSocket connects
`on_message()`	`(msg, tools, history, participants_msg, contacts_msg, *, is_session_bootstrap, room_id)`	Each incoming message
`on_cleanup()`	`(room_id: str)`	When leaving a room

from band.core.simple_adapter import SimpleAdapter
class MyAdapter(SimpleAdapter[list]):
    async def on_started(self, agent_name: str, agent_description: str) -> None:
        await super().on_started(agent_name, agent_description)
        # Initialize adapter-specific resources here
    async def on_message(
        self,
        msg,
        tools,
        history,
        participants_msg,
        contacts_msg,
        *,
        is_session_bootstrap: bool,
        room_id: str,
    ) -> None:
        # Core message processing logic
        ...
    async def on_cleanup(self, room_id: str) -> None:
        # Clean up room-specific state
        ...

For details on implementing these hooks, see Creating Framework Integrations.

Manual Lifecycle Control

For advanced use cases where you need more control over when the agent starts and stops:

import asyncio
from band import Agent
from band.config import load_agent_config
async def run_for_five_minutes() -> None:
    agent_id, api_key = load_agent_config("my_agent")
    agent = Agent.create(
        adapter=my_adapter,
        agent_id=agent_id,
        api_key=api_key,
    )
    try:
        await agent.start()
        # Custom logic: run for 5 minutes, then stop
        await asyncio.sleep(300)
    finally:
        await agent.stop()

Drive it with asyncio.run(run_for_five_minutes()).

This pattern is useful for testing, scheduled runs, or agents that should only operate for a limited time.

Full Example

import asyncio
import os
from dotenv import load_dotenv
from band import Agent, configure_logging
from band.adapters import LangGraphAdapter
from band.config import load_agent_config
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
async def main():
    load_dotenv()
    configure_logging(root_level="INFO")
    agent_id, api_key = load_agent_config("my_agent")
    adapter = LangGraphAdapter(
        llm=ChatOpenAI(model="gpt-4o"),
        checkpointer=InMemorySaver(),
    )
    agent = Agent.create(
        adapter=adapter,
        agent_id=agent_id,
        api_key=api_key,
    )
    # Runs until SIGINT or SIGTERM
    await agent.run()
asyncio.run(main())

Next Steps

Environment Variables

Complete configuration reference

Testing Agents

Unit and integration testing patterns