Agent Communication - AG2

Agents

Agents are the central primitive in AG2 Beta. They maintain state, interact with models, execute tools, and handle user interactions through a clean, conversation-focused API.

Core Communication Primitives

The API is built around two simple methods:

Agent.ask(...) initiates a new turn and returns an AgentReply object.
AgentReply.ask(...) continues an existing conversation, preserving its context and history.

The final result of any turn is safely stored in reply.response; use reply.body for the text.

Basic Communication Example

Here's how easily you can start and continue a conversation:

 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18   from autogen.beta import Agent from autogen.beta.config import OpenAIConfig agent = Agent( "assistant", prompt="You are a helpful assistant.", config=OpenAIConfig("gpt-4o-mini"), ) # Start a new conversation reply = await agent.ask("Give me one sentence about AG2 beta.") print(reply.body) # Continue the exact same conversation context next_turn = await reply.ask("Now make it shorter.") print(next_turn.body) ... 


`<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>`	`<br>from autogen.beta import Agent<br>from autogen.beta.config import OpenAIConfig<br>agent = Agent(<br> "assistant",<br> prompt="You are a helpful assistant.",<br> config=OpenAIConfig("gpt-4o-mini"),<br>)<br># Start a new conversation<br>reply = await agent.ask("Give me one sentence about AG2 beta.")<br>print(reply.body)<br># Continue the exact same conversation context<br>next_turn = await reply.ask("Now make it shorter.")<br>print(next_turn.body)<br>...<br>`

Empowering Agents with Tools

Agents can seamlessly use Python functions as tools. When you provide a list of @tool-decorated functions to an agent, it automatically manages the entire execution lifecycle (model requests to execution and returning results).

 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17   from autogen.beta import Agent, Context, tool from autogen.beta.config import OpenAIConfig @tool async def echo(text: str) -> str: """Useful for repeating exactly what was given.""" return f"echo: {text}" agent = Agent( "assistant", prompt="Use tools when helpful.", config=OpenAIConfig("gpt-4o-mini"), tools=[echo], ) reply = await agent.ask("Call the echo tool with 'hello'.") print(reply.body) 


`<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>`	`<br>from autogen.beta import Agent, Context, tool<br>from autogen.beta.config import OpenAIConfig<br>@tool<br>async def echo(text: str) -> str:<br> """Useful for repeating exactly what was given."""<br> return f"echo: {text}"<br>agent = Agent(<br> "assistant",<br> prompt="Use tools when helpful.",<br> config=OpenAIConfig("gpt-4o-mini"),<br> tools=[echo],<br>)<br>reply = await agent.ask("Call the echo tool with 'hello'.")<br>print(reply.body)<br>`

Adding Human-in-the-Loop (HITL)

Sometimes an agent needs human guidance. You can configure an agent to handle HumanInputRequest events. This is especially effective inside tools where you can get confirmation before taking a sensitive action.

 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26  from autogen.beta import Agent, Context, tool from autogen.beta.config import OpenAIConfig from autogen.beta.events import HumanInputRequest, HumanMessage @tool async def ask_human(context: Context) -> str: # Pauses agent execution to await human input answer = await context.input("Please provide confirmation:") return f"Human said: {answer}" # Define how your application handles the input request def hitl_hook(event: HumanInputRequest) -> HumanMessage: # Here you could block and wait for UI/CLI input. # We return a static response for demonstration. return HumanMessage(content="confirmed") agent = Agent( "assistant", prompt="Use ask_human when needed.", config=OpenAIConfig("gpt-4o-mini"), tools=[ask_human], hitl_hook=hitl_hook, ) reply = await agent.ask("Request confirmation through the tool.") print(reply.body)


`<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br>20<br>21<br>22<br>23<br>24<br>25<br>26<br>`	<br>from autogen.beta import Agent, Context, tool<br>from autogen.beta.config import OpenAIConfig<br>from autogen.beta.events import HumanInputRequest, HumanMessage<br>@tool<br>async def ask_human(context: Context) -> str:<br> # Pauses agent execution to await human input<br> answer = await context.input("Please provide confirmation:")<br> return f"Human said: {answer}"<br># Define how your application handles the input request<br>def hitl_hook(event: HumanInputRequest) -> HumanMessage:<br> # Here you could block and wait for UI/CLI input.<br> # We return a static response for demonstration.<br> return HumanMessage(content="confirmed")<br>agent = Agent(<br> "assistant",<br> prompt="Use ask_human when needed.",<br> config=OpenAIConfig("gpt-4o-mini"),<br> tools=[ask_human],<br> hitl_hook=hitl_hook,<br>)<br>reply = await agent.ask("Request confirmation through the tool.")<br>print(reply.body)<br>

Observing Agent Actions

Need to know exactly what the agent is doing? Pass a MemoryStream when calling ask(). You can attach event subscribers to log actions, save history to a database, or update a user interface in real time.

 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27  from autogen.beta import Agent, Context, MemoryStream from autogen.beta.events import BaseEvent, ModelResponse, ToolCallEvent from autogen.beta.config import OpenAIConfig stream = MemoryStream() # Listen to everything @stream.subscribe() async def on_any_event(event: BaseEvent) -> None: print(f"Event occurred: {event}") # Only listen to specific events @stream.where(ToolCallEvent).subscribe() async def on_tool_call(event: ToolCallEvent) -> None: print("Agent requested tool:", event.name) agent = Agent( "assistant", prompt="You are a helpful assistant.", config=OpenAIConfig("gpt-4o-mini"), }) # Stream captures all events during the ask reply = await agent.ask( "Give me one sentence about AG2 beta.", stream=stream )


`<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br>20<br>21<br>22<br>23<br>24<br>25<br>26<br>27<br>`	<br>from autogen.beta import Agent, Context, MemoryStream<br>from autogen.beta.events import BaseEvent, ModelResponse, ToolCallEvent<br>from autogen.beta.config import OpenAIConfig<br>stream = MemoryStream()<br># Listen to everything<br>@stream.subscribe()<br>async def on_any_event(event: BaseEvent) -> None:<br> print(f"Event occurred: {event}")<br># Only listen to specific events<br>@stream.where(ToolCallEvent).subscribe()<br>async def on_tool_call(event: ToolCallEvent) -> None:<br> print("Agent requested tool:", event.name)<br>agent = Agent(<br> "assistant",<br> prompt="You are a helpful assistant.",<br> config=OpenAIConfig("gpt-4o-mini"),<br>})<br># Stream captures all events during the ask<br>reply = await agent.ask(<br> "Give me one sentence about AG2 beta.",<br> stream=stream<br>)<br>