autogen_core

Core Python API reference for building distributed multi-agent systems

The autogen_core package provides the foundational components for building distributed, event-driven multi-agent systems.

Agent Classes

BaseAgent

Base class for implementing agents with message handling capabilities.

from autogen_core import BaseAgent, AgentId, AgentRuntime, MessageContext

class MyAgent(BaseAgent):
    def __init__(self, agent_id: AgentId, runtime: AgentRuntime):
        super().__init__(agent_id, runtime)

agent_id · AgentId · required

Unique identifier for the agent instance

runtime · AgentRuntime · required

Runtime environment for message passing and agent coordination

id · AgentId

Returns the agent’s unique identifier

RoutedAgent

Decorator-based agent with automatic message routing.

from autogen_core import RoutedAgent, message_handler, rpc
from dataclasses import dataclass

@dataclass
class TaskMessage:
    task: str

class WorkerAgent(RoutedAgent):
    @message_handler
    async def handle_task(self, message: TaskMessage, ctx: MessageContext) -> str:
        return f"Completed: {message.task}"

    @rpc
    async def get_status(self) -> dict:
        return {"status": "ready"}

description · str

Optional description of the agent’s capabilities

Decorators

@message_handler · decorator

Marks a method as a handler for published messages of the parameter type

@rpc · decorator

Marks a method as an RPC endpoint for direct invocation

@event · decorator

Marks a method as an event handler

ClosureAgent

Agent implementation using closure functions for simple message handling.

from autogen_core import ClosureAgent, ClosureContext

async def my_handler(ctx: ClosureContext, message: str) -> str:
    await ctx.send_message("processed", recipient_id)
    return "Done"

agent = ClosureAgent("my_agent", my_handler, runtime)

name · str · required

Agent name identifier

handler · Callable · required

Async function to handle incoming messages

runtime · AgentRuntime · required

Runtime environment

Runtime & Messaging

AgentRuntime

Protocol defining the runtime environment for agent execution and communication.

from autogen_core import AgentRuntime, AgentId, TopicId

async def example(runtime: AgentRuntime):
    # Send direct message
    response = await runtime.send_message(
        message="Hello",
        recipient=AgentId("worker", "default")
    )

    # Publish to topic
    await runtime.publish_message(
        message={"event": "started"},
        topic_id=TopicId("events", "default")
    )

Methods

send_message · async method

Send a message to a specific agent and await response

message · Any · required

The message payload to send

recipient · AgentId · required

Target agent identifier

sender · AgentId | None

Sender agent identifier (None for external)

cancellation_token · CancellationToken | None

Token to cancel the operation

message_id · str | None

Unique message identifier

publish_message · async method

Publish a message to all subscribers of a topic

message · Any · required

The message payload

topic_id · TopicId · required

Topic to publish to

sender · AgentId | None

Sender agent identifier

register_factory · async method

Register an agent factory for dynamic agent creation

type · str | AgentType · required

Unique agent type identifier

agent_factory · Callable · required

Factory function that creates agent instances

expected_class · type[Agent] | None

Expected agent class for type checking

SingleThreadedAgentRuntime

In-process runtime for single-threaded agent execution.

from autogen_core import SingleThreadedAgentRuntime, TypeSubscription

runtime = SingleThreadedAgentRuntime()

# Register agent type
await runtime.register_factory(
    type="worker",
    agent_factory=lambda: WorkerAgent(),
    expected_class=WorkerAgent
)

# Add subscription
await runtime.add_subscription(
    TypeSubscription(topic_type="tasks", agent_type="worker")
)

# Start runtime
runtime.start()

start · method

Start the runtime message processing loop

stop · async method

Gracefully stop the runtime

MessageContext

Context information for message handling.

from autogen_core import MessageContext

@message_handler
async def handle(self, message: str, ctx: MessageContext) -> None:
    print(f"From: {ctx.sender}")
    print(f"Topic: {ctx.topic_id}")
    print(f"Message ID: {ctx.message_id}")

sender · AgentId | None

Identifier of the message sender

topic_id · TopicId | None

Topic the message was published to

message_id · str

Unique message identifier

cancellation_token · CancellationToken

Token for cancelling operations

Identity & Types

AgentId

Unique identifier for an agent instance.

from autogen_core import AgentId

# Create agent ID
agent_id = AgentId(type="worker", key="instance_1")

# Access components
print(agent_id.type)  # "worker"
print(agent_id.key)   # "instance_1"

type · str · required

Agent type name (alphanumeric and underscores)

key · str · required

Instance key (ASCII 32-126 characters)

AgentType

Agent type definition for factory registration.

from autogen_core import AgentType

worker_type = AgentType("worker")

name · str · required

Type name identifier

TopicId

Topic identifier for pub/sub messaging.

from autogen_core import TopicId

topic = TopicId(type="events", source="system")

type · str · required

Topic type

source · str · required

Topic source namespace

AgentProxy

Proxy for remote agent communication.

from autogen_core import AgentProxy

proxy = AgentProxy(agent_id, runtime)
result = await proxy.send("Hello")

Subscriptions

TypeSubscription

Subscribe to messages by exact type match.

from autogen_core import TypeSubscription

subscription = TypeSubscription(
    topic_type="tasks",
    agent_type="worker"
)
await runtime.add_subscription(subscription)

topic_type · str · required

Topic type to subscribe to

agent_type · str · required

Agent type that will receive messages

TypePrefixSubscription

Subscribe to messages matching a type prefix.

from autogen_core import TypePrefixSubscription

subscription = TypePrefixSubscription(
    topic_type_prefix="task.",
    agent_type="worker"
)

topic_type_prefix · str · required

Topic type prefix to match

agent_type · str · required

Agent type to receive matching messages

DefaultSubscription

Subscribe to all messages in a namespace.

from autogen_core import DefaultSubscription

subscription = DefaultSubscription(
    agent_type="logger"
)

Serialization & Components

MessageSerializer

Protocol for custom message serialization.

from autogen_core import MessageSerializer

class MySerializer(MessageSerializer):
    def serialize(self, message: Any) -> bytes:
        return json.dumps(message).encode()

    def deserialize(self, data: bytes, type_name: str) -> Any:
        return json.loads(data.decode())

Component System

Declarative component configuration.

from autogen_core import Component, ComponentModel

@Component
class MyComponent:
    def __init__(self, config: dict):
        self.value = config["value"]

    def to_config(self) -> ComponentModel:
        return ComponentModel(
            component_type=self.__class__.__name__,
            config={"value": self.value}
        )

Component · decorator

Marks a class as a configurable component

ComponentModel · class

Serializable component configuration

component_type · str · required

Component type identifier

config · dict · required

Component configuration data

Image

Image data container for multimodal messages.

from autogen_core import Image

# From bytes
image = Image.from_bytes(image_bytes, format="png")

# From file
image = Image.from_file("photo.jpg")

# From URL
image = Image.from_url("https://example.com/image.png")

from_bytes · classmethod

Create image from byte data

from_file · classmethod

Create image from file path

from_url · classmethod

Create image from URL

State Management

CacheStore

Protocol for agent state persistence.

from autogen_core import CacheStore

class CustomStore(CacheStore):
    async def get(self, key: str) -> Any:
        # Retrieve value
        pass

    async def set(self, key: str, value: Any) -> None:
        # Store value
        pass

InMemoryStore

In-memory cache store implementation.

from autogen_core import InMemoryStore

store = InMemoryStore()
await store.set("key", "value")
result = await store.get("key")

Intervention & Control

InterventionHandler

Protocol for intercepting and modifying messages.

from autogen_core import InterventionHandler, MessageContext

class ApprovalHandler(InterventionHandler):
    async def on_send(self, message: Any, ctx: MessageContext) -> Any:
        # Review and modify message before sending
        if needs_approval(message):
            approved = await get_approval(message)
            if not approved:
                from autogen_core import DropMessage
                raise DropMessage("Not approved")
        return message

DropMessage

Exception to prevent message delivery.

from autogen_core import DropMessage

raise DropMessage("Message rejected by policy")

CancellationToken

Token for cancelling long-running operations.

from autogen_core import CancellationToken

token = CancellationToken()

# Cancel after timeout
token.cancel_after(10.0)

# Check if cancelled
if token.is_cancelled:
    return

Telemetry

Tracing

OpenTelemetry integration for distributed tracing.

from autogen_core import (
    trace_create_agent_span,
    trace_invoke_agent_span,
    trace_tool_span
)

# Trace agent creation
with trace_create_agent_span(agent_type="worker"):
    agent = WorkerAgent()

# Trace agent invocation
with trace_invoke_agent_span(agent_id, message_type="Task"):
    result = await agent.handle(task)

# Trace tool execution
with trace_tool_span(tool_name="calculator"):
    result = calculator.add(1, 2)

Logging

Structured logging configuration.

from autogen_core import ROOT_LOGGER_NAME, EVENT_LOGGER_NAME, TRACE_LOGGER_NAME
import logging

# Configure root logger
logging.getLogger(ROOT_LOGGER_NAME).setLevel(logging.INFO)

# Configure event logger for structured events
event_logger = logging.getLogger(EVENT_LOGGER_NAME)

# Configure trace logger for development debugging
trace_logger = logging.getLogger(TRACE_LOGGER_NAME)

ROOT_LOGGER_NAME · str

Name of the root logger: "autogen_core"

EVENT_LOGGER_NAME · str

Name for structured event logging

TRACE_LOGGER_NAME · str

Name for development trace logging

Types & Protocols

FunctionCall

Represents a tool or function call.

from autogen_core import FunctionCall

call = FunctionCall(
    id="call_123",
    name="calculator",
    arguments={"a": 1, "b": 2}
)

id · str · required

Unique call identifier

name · str · required

Function or tool name

arguments · dict · required

Function arguments

UnknownPayload

Represents a message with unknown type.

from autogen_core import UnknownPayload

# Raised when deserializer encounters unknown type

Constants

JSON_DATA_CONTENT_TYPE = "application/json"
PROTOBUF_DATA_CONTENT_TYPE = "application/protobuf"

See Also

• autogen_agentchat - High-level chat agent framework
• autogen_ext - Extensions and integrations