Agent Lifecycle

This guide covers the operational lifecycle of a Thenvoi 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()`	Connects to platform, 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.

Startup Sequence

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

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

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

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

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

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 platform connection is established
`on_message()`	`(msg, tools, history, participants_msg, *, is_session_bootstrap, room_id)`	Each incoming message
`on_cleanup()`	`(room_id: str)`	When leaving a room

1 from thenvoi.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         *,
15         is_session_bootstrap: bool,
16         room_id: str,
17     ) -> None:
18         # Core message processing logic
19         ...
20 
21     async def on_cleanup(self, room_id: str) -> None:
22         # Clean up room-specific state
23         ...

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 from thenvoi import Agent
3 
4 agent = Agent.create(
5     adapter=my_adapter,
6     agent_id=agent_id,
7     api_key=api_key,
8 )
9 
10 try:
11     await agent.start()
12 
13     # Custom logic: run for 5 minutes, then stop
14     await asyncio.sleep(300)
15 
16 finally:
17     await agent.stop()

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 logging
3 import os
4 from dotenv import load_dotenv
5 from thenvoi import Agent
6 from thenvoi.adapters import LangGraphAdapter
7 from thenvoi.config import load_agent_config
8 from langchain_openai import ChatOpenAI
9 from langgraph.checkpoint.memory import InMemorySaver
10 
11 logging.basicConfig(level=logging.INFO)
12 
13 async def main():
14     load_dotenv()
15     agent_id, api_key = load_agent_config("my_agent")
16 
17     adapter = LangGraphAdapter(
18         llm=ChatOpenAI(model="gpt-4o"),
19         checkpointer=InMemorySaver(),
20     )
21 
22     agent = Agent.create(
23         adapter=adapter,
24         agent_id=agent_id,
25         api_key=api_key,
26         ws_url=os.getenv("THENVOI_WS_URL"),
27         rest_url=os.getenv("THENVOI_REST_URL"),
28     )
29 
30     # Runs until SIGINT or SIGTERM
31     await agent.run()
32 
33 asyncio.run(main())

Next Steps

Environment Variables

Complete configuration reference

Testing Agents

Unit and integration testing patterns