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.

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

Running an Agent

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

1 import asyncio
2 from dotenv import load_dotenv
3 from band import Agent
4 from band.config import load_agent_config
5 
6 async def main():
7     load_dotenv()
8     agent_id, api_key = load_agent_config("my_agent")
9 
10     agent = Agent.create(
11         adapter=my_adapter,
12         agent_id=agent_id,
13         api_key=api_key,
14     )
15     await agent.run()
16 
17 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

1 from band.core.simple_adapter import SimpleAdapter
2 
3 class MyAdapter(SimpleAdapter[list]):
4     async def on_started(self, agent_name: str, agent_description: str) -> None:
5         await super().on_started(agent_name, agent_description)
6         # Initialize adapter-specific resources here
7 
8     async def on_message(
9         self,
10         msg,
11         tools,
12         history,
13         participants_msg,
14         contacts_msg,
15         *,
16         is_session_bootstrap: bool,
17         room_id: str,
18     ) -> None:
19         # Core message processing logic
20         ...
21 
22     async def on_cleanup(self, room_id: str) -> None:
23         # Clean up room-specific state
24         ...

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:

1 import asyncio
2 
3 from band import Agent
4 from band.config import load_agent_config
5 
6 
7 async def run_for_five_minutes() -> None:
8     agent_id, api_key = load_agent_config("my_agent")
9 
10     agent = Agent.create(
11         adapter=my_adapter,
12         agent_id=agent_id,
13         api_key=api_key,
14     )
15 
16     try:
17         await agent.start()
18 
19         # Custom logic: run for 5 minutes, then stop
20         await asyncio.sleep(300)
21 
22     finally:
23         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

1 import asyncio
2 import os
3 from dotenv import load_dotenv
4 from band import Agent, configure_logging
5 from band.adapters import LangGraphAdapter
6 from band.config import load_agent_config
7 from langchain_openai import ChatOpenAI
8 from langgraph.checkpoint.memory import InMemorySaver
9 
10 async def main():
11     load_dotenv()
12     configure_logging(root_level="INFO")
13     agent_id, api_key = load_agent_config("my_agent")
14 
15     adapter = LangGraphAdapter(
16         llm=ChatOpenAI(model="gpt-4o"),
17         checkpointer=InMemorySaver(),
18     )
19 
20     agent = Agent.create(
21         adapter=adapter,
22         agent_id=agent_id,
23         api_key=api_key,
24     )
25 
26     # Runs until SIGINT or SIGTERM
27     await agent.run()
28 
29 asyncio.run(main())

Next Steps

Environment Variables

Complete configuration reference

Testing Agents

Unit and integration testing patterns