Core Concepts
Context
The runtime environment that wires channels, agents, and state together.
A Context is the runtime container for Zgent. It manages:
- Receivers — message sources (channels that produce input)
- Senders — message destinations (channels that consume output)
- Agent definitions — templates for creating agent instances
- Agent instances — live agents handling sessions
Creating a context
from zgent import Context, AgentDefinition
ctx = Context(
id="my-project",
cwd="/path/to/workspace",
receivers=[input_channel],
senders=[output_channel],
definitions=[agent_def],
)| Parameter | Description |
|---|---|
id | Unique identifier for this context |
cwd | Working directory for file operations |
receivers | List of channels that produce input messages |
senders | List of channels that receive output messages |
definitions | List of AgentDefinition templates |
snapshot_store | Optional store for persisting agent state |
Running the context
Call serve() to start the message loop. The context listens on all receivers, dispatches messages to agents, and forwards output to senders.
import asyncio
async def main():
await input_channel.connect()
await output_channel.connect()
task = asyncio.create_task(ctx.serve())
# ... interact with channels ...
await ctx.stop()
task.cancel()Registering definitions at runtime
You can add agent definitions after construction:
ctx = Context(id="dynamic", cwd="/tmp")
ctx.register(agent_def_a)
ctx.register(agent_def_b)State management
Each agent instance maintains its own AgentContext with conversation history, token usage, and snapshots. The context coordinates instance creation and cleanup.
When a snapshot_store is provided, agent state is persisted across restarts:
from zgent import AgentContext, AgentSnapshot, SnapshotStore
snapshots: dict[str, AgentSnapshot] = {}
async def load_snapshot(dispatch_key: str) -> AgentSnapshot | None:
return snapshots.get(dispatch_key)
async def save_snapshot(ctx: AgentContext, snapshot: AgentSnapshot) -> None:
snapshots[ctx.session_id] = snapshot
store = SnapshotStore(load=load_snapshot, save=save_snapshot)
ctx = Context(id="persistent", cwd="/tmp", snapshot_store=store)