autogen_core_single_agent_message_handler_quickstart.py

1import asyncio
2from autogen_core import AgentId, MessageContext, SingleThreadedAgentRuntime
3from autogen_core import RoutedAgent, message_handler
4from dataclasses import dataclass
5
6@dataclass
7class Message:
8    content: str
9
10class MyAgent(RoutedAgent):
11    def __init__(self, description: str) -> None:
12        super().__init__(description)
13
14    @message_handler
15    async def handle_message(self, message: Message, ctx: MessageContext) -> Message:
16        print(f"Received message: {message.content}")
17        return Message(content=f"Hello, {message.content}!")
18
19async def main() -> None:
20    # 1. Create a runtime
21    runtime = SingleThreadedAgentRuntime()
22
23    # 2. Register an agent by providing a factory function
24    await MyAgent.register(
25        runtime, 
26        "my_agent", 
27        lambda: MyAgent("A simple agent")
28    )
29
30    # 3. Start the runtime
31    runtime.start()
32
33    # 4. Send a message to the agent and wait for a response
34    response = await runtime.send_message(
35        Message("World"), 
36        AgentId("my_agent", "default")
37    )
38    print(f"Response: {response.content}")
39
40    # 5. Stop the runtime
41    await runtime.stop()
42
43if __name__ == "__main__":
44    asyncio.run(main())