Skip to main content
The Agent Runtime is the execution environment that manages agent lifecycle, message routing, and system-level operations. It provides the infrastructure for agents to send messages, publish events, and maintain state.

​
AgentRuntime Protocol

The AgentRuntime protocol defines the interface all runtime implementations must provide: 
from autogen_core import AgentRuntime, AgentId, TopicId, Agent, Subscription
from typing import Any, Awaitable, Callable

class AgentRuntime(Protocol):
    async def send_message(
        self,
        message: Any,
        recipient: AgentId,
        *,
        sender: AgentId | None = None,
        cancellation_token: CancellationToken | None = None,
        message_id: str | None = None,
    ) -> Any:
        """Send a message to an agent and get a response."""
        ...
    
    async def publish_message(
        self,
        message: Any,
        topic_id: TopicId,
        *,
        sender: AgentId | None = None,
        cancellation_token: CancellationToken | None = None,
        message_id: str | None = None,
    ) -> None:
        """Publish a message to all subscribers of a topic."""
        ...
    
    async def register_factory(
        self,
        type: str | AgentType,
        agent_factory: Callable[[], T | Awaitable[T]],
        *,
        expected_class: type[T] | None = None,
    ) -> AgentType:
        """Register an agent factory for on-demand instantiation."""
        ...
    
    async def register_agent_instance(
        self,
        agent_instance: Agent,
        agent_id: AgentId,
    ) -> AgentId:
        """Register a specific agent instance."""
        ...
    
    async def add_subscription(self, subscription: Subscription) -> None:
        """Add a subscription for message routing."""
        ...
    
    async def save_state(self) -> Mapping[str, Any]:
        """Save runtime and all agent states."""
        ...
    
    async def load_state(self, state: Mapping[str, Any]) -> None:
        """Restore runtime and all agent states."""
        ...

​
SingleThreadedAgentRuntime

SingleThreadedAgentRuntime is a development-focused runtime that processes all messages in a single asyncio event loop.
SingleThreadedAgentRuntime is designed for development and standalone applications. For production deployments with high throughput or distributed requirements, use the distributed runtime.

​
Initialization

from autogen_core import SingleThreadedAgentRuntime
from opentelemetry.trace import TracerProvider

runtime = SingleThreadedAgentRuntime(
    intervention_handlers=None,  # Optional message interceptors
    tracer_provider=None,        # Optional OpenTelemetry provider
    ignore_unhandled_exceptions=True  # Ignore exceptions in event handlers
)
Parameters:
​
intervention_handlers
List[InterventionHandler]
default:"None"
List of handlers that can intercept and modify messages before delivery. Useful for logging, validation, or message transformation.
​
tracer_provider
TracerProvider
default:"None"
OpenTelemetry tracer provider for distributed tracing. Set AUTOGEN_DISABLE_RUNTIME_TRACING=true to disable.
​
ignore_unhandled_exceptions
bool
default:"True"
If True, exceptions in event handlers won’t stop the runtime. Exceptions are raised on next process_next() or stop() call. RPC handler exceptions are always propagated.

​
Lifecycle Management

​
Starting the Runtime

import asyncio
from autogen_core import SingleThreadedAgentRuntime

async def main():
    runtime = SingleThreadedAgentRuntime()
    
    # Register agents...
    
    # Start background message processing
    runtime.start()
    
    # Runtime is now processing messages
    # Your application logic here...
    
    # Stop when done
    await runtime.stop_when_idle()

asyncio.run(main())

​
Stopping the Runtime

Three ways to stop the runtime: 
# 1. Stop immediately (discards queued messages)
await runtime.stop()

# 2. Stop when message queue is empty (recommended)
await runtime.stop_when_idle()

# 3. Stop when condition is met (legacy, not recommended)
await runtime.stop_when(lambda: task_complete)
Use stop_when_idle() in most cases. It ensures all queued messages are processed before shutdown.

​
Cleanup

# Close runtime and all agents
await runtime.close()
This calls stop() if the runtime is running, then calls close() on all instantiated agents.

​
Agent Registration

​
Factory Registration

Register a factory function that creates agents on-demand: 
from autogen_core import SingleThreadedAgentRuntime, AgentType

runtime = SingleThreadedAgentRuntime()

# Using BaseAgent.register() (recommended)
await MyAgent.register(
    runtime=runtime,
    type="my_agent",
    factory=lambda: MyAgent("Agent description")
)

# Using runtime.register_factory() (low-level)
agent_type = await runtime.register_factory(
    type="my_agent",
    agent_factory=lambda: MyAgent("Agent description"),
    expected_class=MyAgent  # Optional runtime validation
)
Factory function signature: 
# Zero-argument factory (recommended)
def factory() -> Agent:
    return MyAgent()

# Access runtime and ID via context
from autogen_core import AgentInstantiationContext

def factory() -> Agent:
    runtime = AgentInstantiationContext.current_runtime()
    agent_id = AgentInstantiationContext.current_agent_id()
    return MyAgent(runtime, agent_id)
Two-argument factories factory(runtime: AgentRuntime, agent_id: AgentId) are deprecated. Use AgentInstantiationContext instead.

​
Instance Registration

Register a pre-created agent instance: 
from autogen_core import AgentId

agent = MyAgent("My agent")

# Using BaseAgent.register_instance() (recommended)
await agent.register_instance(
    runtime=runtime,
    agent_id=AgentId("my_agent", "instance-1")
)

# Using runtime.register_agent_instance() (low-level)
await runtime.register_agent_instance(
    agent_instance=agent,
    agent_id=AgentId("my_agent", "instance-1")
)
All instances of the same agent type must be of the same class. Mixing factories and instances for the same type is not allowed.

​
Message Operations

​
Sending Direct Messages

from autogen_core import AgentId, CancellationToken
from dataclasses import dataclass

@dataclass
class TaskRequest:
    task_id: str

# Send message and await response
response = await runtime.send_message(
    message=TaskRequest(task_id="123"),
    recipient=AgentId("worker", "instance-1"),
    sender=AgentId("coordinator", "default"),  # Optional
    cancellation_token=CancellationToken(),  # Optional
    message_id="unique-msg-id"  # Optional, auto-generated if None
)
Parameters:
​
message
Any
required
The message object to send. Must be serializable.
​
recipient
AgentId
required
The agent that should receive and process the message.
​
sender
AgentId | None
default:"None"
The sending agent’s ID. Should only be None for external (non-agent) senders.
​
cancellation_token
CancellationToken | None
default:"None"
Token to cancel the operation if needed.
​
message_id
str | None
default:"None"
Unique message identifier. Auto-generated UUID if not provided.

​
Publishing to Topics

from autogen_core import TopicId

@dataclass
class StatusUpdate:
    status: str

# Publish to all subscribers
await runtime.publish_message(
    message=StatusUpdate(status="processing"),
    topic_id=TopicId(type="status", source="system"),
    sender=AgentId("worker", "instance-1")  # Optional
)
publish_message() is fire-and-forget. It doesn’t wait for responses or confirm delivery to subscribers.

​
Subscription Management

from autogen_core import TypeSubscription, DefaultSubscription

# Subscribe agent to topic
await runtime.add_subscription(
    TypeSubscription(
        topic_type="status",
        agent_type="logger"
    )
)

# Remove subscription by ID
await runtime.remove_subscription(subscription_id="sub-uuid")
See Message Passing for details on subscription types.

​
State Persistence

​
Save State

# Save all agent states
state = await runtime.save_state()
print(state)  # {'agent_type/key': {'counter': 42, ...}, ...}

# Save to file
import json
with open('runtime_state.json', 'w') as f:
    json.dump(state, f)

​
Load State

import json

# Load from file
with open('runtime_state.json', 'r') as f:
    state = json.load(f)

# Restore state
await runtime.load_state(state)

​
Per-Agent State Operations

from autogen_core import AgentId

agent_id = AgentId("worker", "instance-1")

# Get agent metadata
metadata = await runtime.agent_metadata(agent_id)
print(metadata.type, metadata.key, metadata.description)

# Save single agent state
agent_state = await runtime.agent_save_state(agent_id)

# Load single agent state
await runtime.agent_load_state(agent_id, agent_state)

​
Advanced Operations

​
Manual Message Processing

# Process messages without start()
while has_more_work:
    await runtime.process_next()  # Process one message from queue
process_next() raises unhandled exceptions. Cannot be called again after an exception is raised. Use start() for automatic message processing.

​
Accessing Agent Instances

from autogen_core import AgentId

# Get underlying agent instance (discouraged)
try:
    agent = await runtime.try_get_underlying_agent_instance(
        id=AgentId("worker", "instance-1"),
        type=MyAgent  # Expected type
    )
    # Direct access to agent
    print(agent.internal_state)
except LookupError:
    print("Agent not found")
except TypeError:
    print("Agent is wrong type")
Accessing agent instances directly breaks the Actor model. Use only for debugging or special cases.

​
Agent Lookup

from autogen_core import AgentType

# Get AgentId for lazy or eager instantiation
agent_id = await runtime.get(
    AgentType("worker"),
    key="instance-1",
    lazy=True  # Don't create agent yet
)

# Or from existing AgentId
agent_id = await runtime.get(existing_agent_id)

​
Message Serialization

from autogen_core import MessageSerializer
from typing import Any

class CustomSerializer(MessageSerializer[MyMessage]):
    @property
    def data_content_type(self) -> str:
        return "application/x-custom"
    
    def serialize(self, message: MyMessage) -> bytes:
        return json.dumps(message.__dict__).encode()
    
    def deserialize(self, data: bytes) -> MyMessage:
        return MyMessage(**json.loads(data))

# Register serializer
runtime.add_message_serializer(CustomSerializer())

​
Complete Example

import asyncio
from dataclasses import dataclass
from autogen_core import (
    SingleThreadedAgentRuntime,
    RoutedAgent,
    MessageContext,
    AgentId,
    TopicId,
    TypeSubscription,
    rpc,
    event
)

@dataclass
class WorkRequest:
    work_id: str
    data: str

@dataclass
class WorkResponse:
    work_id: str
    result: str

@dataclass
class ProgressEvent:
    work_id: str
    progress: float

class Worker(RoutedAgent):
    def __init__(self):
        super().__init__("Worker agent")
    
    @rpc
    async def process_work(self, message: WorkRequest, ctx: MessageContext) -> WorkResponse:
        # Publish progress updates
        for progress in [0.25, 0.5, 0.75, 1.0]:
            await self.publish_message(
                ProgressEvent(work_id=message.work_id, progress=progress),
                TopicId(type="progress", source=message.work_id)
            )
            await asyncio.sleep(0.1)
        
        result = f"Processed: {message.data}"
        return WorkResponse(work_id=message.work_id, result=result)

class Monitor(RoutedAgent):
    def __init__(self):
        super().__init__("Monitor agent")
        self.progress_log = []
    
    @event
    async def track_progress(self, message: ProgressEvent, ctx: MessageContext) -> None:
        self.progress_log.append((message.work_id, message.progress))
        print(f"Progress: {message.work_id} - {message.progress*100}%")

async def main():
    # Create and configure runtime
    runtime = SingleThreadedAgentRuntime()
    
    # Register agents
    await Worker.register(runtime, "worker", lambda: Worker())
    await Monitor.register(runtime, "monitor", lambda: Monitor())
    
    # Add subscription for monitor to receive progress events
    await runtime.add_subscription(
        TypeSubscription(topic_type="progress", agent_type="monitor")
    )
    
    # Start runtime
    runtime.start()
    
    try:
        # Send work request
        response = await runtime.send_message(
            WorkRequest(work_id="task-1", data="sample data"),
            recipient=AgentId("worker", "default")
        )
        print(f"Result: {response.result}")
        
        # Give time for progress events to process
        await asyncio.sleep(0.5)
        
    finally:
        # Clean shutdown
        await runtime.stop_when_idle()
        await runtime.close()

if __name__ == "__main__":
    asyncio.run(main())

​
Runtime Monitoring

# Check message queue size
queue_size = runtime.unprocessed_messages_count
print(f"Messages pending: {queue_size}")

​
Next Steps

Message Passing

Learn about messages, contexts, topics, and subscriptions

Event-Driven Architecture

Understand event handlers and message routing

Distributed Runtime

Scale across processes and languages

Core Overview

Return to Core API overview