autogen_core

from autogen_core import BaseAgent, AgentId, AgentRuntime, MessageContext

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

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"}

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)

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")
    )

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()

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}")

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"

from autogen_core import AgentType

worker_type = AgentType("worker")

from autogen_core import TopicId

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

from autogen_core import AgentProxy

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

from autogen_core import TypeSubscription

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

from autogen_core import TypePrefixSubscription

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

from autogen_core import DefaultSubscription

subscription = DefaultSubscription(
    agent_type="logger"
)

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())

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}
        )

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 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

from autogen_core import InMemoryStore

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

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

from autogen_core import DropMessage

raise DropMessage("Message rejected by policy")

from autogen_core import CancellationToken

token = CancellationToken()

# Cancel after timeout
token.cancel_after(10.0)

# Check if cancelled
if token.is_cancelled:
    return

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)

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)

from autogen_core import FunctionCall

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

from autogen_core import UnknownPayload

# Raised when deserializer encounters unknown type