Skip to content

Commit b00465d

Browse files
Shubham-Kumar-2000eavanvalkenburg
Shubham-Kumar-2000
and
eavanvalkenburg
authored
Python: feat: Add Agent Framework to A2A bridge support (#2403)
* feat: Add Agent Framework to A2A bridge support - Implement A2A event adapter for converting agent messages to A2A protocol - Add A2A execution context for managing agent execution state - Implement A2A executor for running agents in A2A environment - Add comprehensive unit tests for event adapter, execution context, and executor - Update agent framework core A2A module exports and type stubs - Integrate thread management utilities for async execution - Add getting started sample for A2A agent framework integration - Update dependencies in uv.lock This integration enables agent framework agents to communicate and execute within the A2A (Agent to Agent) infrastructure. * fix: Update references from agent_thread_storage to _agent_thread_storage in A2A executor tests * Refactor A2A agent framework and improve code structure - Reordered imports in various files for consistency and clarity. - Updated `__all__` definitions to maintain a consistent order across modules. - Simplified method signatures by removing unnecessary line breaks. - Enhanced readability by adjusting formatting in several sections. - Removed redundant comments and example scenarios in the execution context. - Improved handling of agent messages in the event adapter. - Added type hints for better clarity and type checking. - Cleaned up test cases for better organization and readability. * fix: Lint fix new line added * test: Add unit tests for AgentThreadStorage and InMemoryAgentThreadStorage * refactor: Update type hints to use new syntax for Union and List * fix: Validate RequestContext for context_id and message before execution * Refactor tests and remove A2aExecutionContext references - Deleted the test file for A2aExecutionContext as it is no longer needed. - Updated A2aExecutor tests to remove dependencies on A2aExecutionContext and adjusted method calls accordingly. - Modified event adapter tests to use ChatMessage instead of AgentRunResponseUpdate. - Removed A2aExecutionContext from imports in agent_framework.a2a module and updated type hints accordingly. * Refactor A2AExecutor tests and remove event adapter - Updated test cases to use A2AExecutor instead of A2aExecutor for consistency. - Removed mock_event_adapter fixture and related tests as A2aEventAdapter is deprecated. - Consolidated event handling tests into TestA2AExecutorEventAdapter. - Adjusted imports in various files to reflect the removal of deprecated components. - Ensured all references to A2aExecutor are updated to A2AExecutor across the codebase. * refactor: Remove AgentThreadStorage and InMemoryAgentThreadStorage classes from threads and tests * feat: A2AExecutor to have its own override able save and get threads methods for persistent storage. * fix: linter bugs * removed unnecessary changes form core package * new line added * Refactor A2AExecutor tests and update imports - Consolidated mock agent fixtures in test_a2a_executor.py to simplify agent mocking. - Removed redundant tests related to thread storage and agent types, focusing on A2AExecutor's core functionality. - Updated test assertions to reflect changes in message handling with new Message and Content classes. - Enhanced integration tests to ensure compatibility with the new agent framework structure. - Added A2AExecutor to the module exports in __init__.py and __init__.pyi for better accessibility. * Update A2A documentation: enhance usage examples for A2AAgent and A2AExecutor * Updated uv lock * Fix metadata assertion in TestA2AExecutorHandleEvents and reorder load_dotenv call in agent_framework_to_a2a.py * Update agent card configuration: add default input and output modes, and fix agent creation method * Fix assertion for metadata in TestA2AExecutorHandleEvents * Fix formatting issues in TestA2AExecutorExecute and TestA2AExecutorIntegration * Enhance A2AExecutor documentation with examples and clarify agent execution process * Revert uv lock to main * Refactor A2AExecutor: Improve formatting and streamline constructor parameters * Apply suggestions from code review Co-authored-by: Eduard van Valkenburg <[email protected]> * Refactor A2AExecutor to use SupportsAgentRun and enhance logging; update agent framework sample for flight and hotel booking capabilities * Enhance A2AExecutor with streaming support and custom run arguments; update tests for initialization and execution scenarios * Enhance A2AExecutor event handling with streamed artifact tracking; update tests for new behavior * Refactor A2AExecutor to enforce type hints for stream and run_kwargs attributes * Refactor A2AExecutor and tests: replace AsyncMock with MagicMock for response stream handling; clean up imports in agent_framework_to_a2a.py * refactor: streamline imports and improve code readability across multiple files * feat: enhance A2AExecutor cancel method with context validation and fixed review comments * feat: implement get_uri_data utility function for extracting base64 data from data URIs and update references * fix: update import path for get_uri_data utility function in A2AExecutor and A2AAgent * fix: correct error message handling in A2AExecutor and update test assertions --------- Co-authored-by: Eduard van Valkenburg <[email protected]>
1 parent 4adfd24 commit b00465d

13 files changed

Lines changed: 1407 additions & 27 deletions

File tree

‎python/packages/a2a/AGENTS.md‎

Lines changed: 32 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,20 +4,48 @@ Agent-to-Agent (A2A) protocol support for inter-agent communication.
44

55
## Main Classes
66

7-
- **`A2AAgent`** - Agent wrapper that exposes an agent via the A2A protocol
7+
- **`A2AAgent`** - Client to connect to remote A2A-compliant agents.
8+
- **`A2AExecutor`** - Bridge to expose Agent Framework agents via the A2A protocol.
89

910
## Usage
1011

12+
### A2AAgent (Client)
13+
1114
```python
1215
from agent_framework.a2a import A2AAgent
1316

14-
a2a_agent = A2AAgent(agent=my_agent)
17+
# Connect to a remote A2A agent
18+
a2a_agent = A2AAgent(url="http://remote-agent/a2a")
19+
response = await a2a_agent.run("Hello!")
20+
```
21+
22+
### A2AExecutor (Server/Bridge)
23+
24+
```python
25+
from agent_framework.a2a import A2AExecutor
26+
from a2a.server.apps import A2AStarletteApplication
27+
from a2a.server.request_handlers import DefaultRequestHandler
28+
from a2a.server.tasks import InMemoryTaskStore
29+
30+
# Create an A2A executor for your agent
31+
executor = A2AExecutor(agent=my_agent)
32+
33+
# Set up the request handler and server application
34+
request_handler = DefaultRequestHandler(
35+
agent_executor=executor,
36+
task_store=InMemoryTaskStore(),
37+
)
38+
39+
app = A2AStarletteApplication(
40+
agent_card=my_agent_card,
41+
http_handler=request_handler,
42+
).build()
1543
```
1644

1745
## Import Path
1846

1947
```python
20-
from agent_framework.a2a import A2AAgent
48+
from agent_framework.a2a import A2AAgent, A2AExecutor
2149
# or directly:
22-
from agent_framework_a2a import A2AAgent
50+
from agent_framework_a2a import A2AAgent, A2AExecutor
2351
```

‎python/packages/a2a/README.md‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,49 @@ pip install agent-framework-a2a --pre
1010

1111
The A2A agent integration enables communication with remote A2A-compliant agents using the standardized A2A protocol. This allows your Agent Framework applications to connect to agents running on different platforms, languages, or services.
1212

13+
### A2AAgent (Client)
14+
15+
The `A2AAgent` class is a client that wraps an A2A Client to connect the Agent Framework with external A2A-compliant agents.
16+
17+
```python
18+
from agent_framework.a2a import A2AAgent
19+
20+
# Connect to a remote A2A agent
21+
a2a_agent = A2AAgent(url="http://remote-agent/a2a")
22+
response = await a2a_agent.run("Hello!")
23+
```
24+
25+
### A2AExecutor (Hosting)
26+
27+
The `A2AExecutor` class bridges local AI agents built with the `agent_framework` library to the A2A protocol, allowing them to be hosted and accessed by other A2A-compliant clients.
28+
29+
```python
30+
from agent_framework.a2a import A2AExecutor
31+
from a2a.server.apps import A2AStarletteApplication
32+
from a2a.server.request_handlers import DefaultRequestHandler
33+
from a2a.server.tasks import InMemoryTaskStore
34+
35+
# Create an A2A executor for your agent
36+
executor = A2AExecutor(agent=my_agent)
37+
38+
# Set up the request handler and server application
39+
request_handler = DefaultRequestHandler(
40+
agent_executor=executor,
41+
task_store=InMemoryTaskStore(),
42+
)
43+
44+
app = A2AStarletteApplication(
45+
agent_card=my_agent_card,
46+
http_handler=request_handler,
47+
).build()
48+
```
49+
1350
### Basic Usage Example
1451

1552
See the [A2A agent examples](../../samples/04-hosting/a2a/) which demonstrate:
1653

1754
- Connecting to remote A2A agents
55+
- Hosting local agents via A2A protocol
1856
- Sending messages and receiving responses
1957
- Handling different content types (text, files, data)
2058
- Streaming responses and real-time interaction

‎python/packages/a2a/agent_framework_a2a/__init__.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import importlib.metadata
44

5+
from ._a2a_executor import A2AExecutor
56
from ._agent import A2AAgent, A2AContinuationToken
67

78
try:
@@ -12,5 +13,6 @@
1213
__all__ = [
1314
"A2AAgent",
1415
"A2AContinuationToken",
16+
"A2AExecutor",
1517
"__version__",
1618
]

‎python/packages/a2a/agent_framework_a2a/_a2a_executor.py‎

Lines changed: 275 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,275 @@
1+
# Copyright (c) Microsoft. All rights reserved.
2+
3+
import logging
4+
from asyncio import CancelledError
5+
from collections.abc import Mapping
6+
from functools import partial
7+
from typing import Any
8+
9+
from a2a.server.agent_execution import AgentExecutor, RequestContext
10+
from a2a.server.events import EventQueue
11+
from a2a.server.tasks import TaskUpdater
12+
from a2a.types import FilePart, FileWithBytes, FileWithUri, Part, TaskState, TextPart
13+
from a2a.utils import new_task
14+
from agent_framework import (
15+
AgentResponseUpdate,
16+
AgentSession,
17+
Message,
18+
SupportsAgentRun,
19+
)
20+
from typing_extensions import override
21+
22+
from agent_framework_a2a._utils import get_uri_data
23+
24+
logger = logging.getLogger("agent_framework.a2a")
25+
26+
27+
class A2AExecutor(AgentExecutor):
28+
"""Execute AI agents using the A2A (Agent-to-Agent) protocol.
29+
30+
The A2AExecutor bridges AI agents built with the agent_framework library and the A2A protocol,
31+
enabling structured agent execution with event-driven communication. It handles execution
32+
contexts, delegates history management to the agent's session, and converts agent
33+
responses into A2A protocol events.
34+
35+
The executor supports executing an Agent or WorkflowAgent. It provides comprehensive
36+
error handling with task status updates and supports various content types including text,
37+
binary data, and URI-based content.
38+
39+
Example:
40+
.. code-block:: python
41+
42+
from a2a.server.apps import A2AStarletteApplication
43+
from a2a.server.request_handlers import DefaultRequestHandler
44+
from a2a.server.tasks import InMemoryTaskStore
45+
from a2a.types import AgentCapabilities, AgentCard
46+
from agent_framework.a2a import A2AExecutor
47+
from agent_framework.openai import OpenAIResponsesClient
48+
49+
public_agent_card = AgentCard(
50+
name="Food Agent",
51+
description="A simple agent that provides food-related information.",
52+
url="http://localhost:9999/",
53+
version="1.0.0",
54+
defaultInputModes=["text"],
55+
defaultOutputModes=["text"],
56+
capabilities=AgentCapabilities(streaming=True),
57+
skills=[],
58+
)
59+
60+
# Create an agent
61+
agent = OpenAIResponsesClient().as_agent(
62+
name="Food Agent",
63+
instructions="A simple agent that provides food-related information.",
64+
)
65+
66+
# Set up the A2A server with the A2AExecutor enabled for streaming
67+
# and passing custom keyword arguments to the agent's run method.
68+
request_handler = DefaultRequestHandler(
69+
agent_executor=A2AExecutor(agent, stream=True, run_kwargs={"client_kwargs": {"max_tokens": 500}}),
70+
task_store=InMemoryTaskStore(),
71+
)
72+
73+
server = A2AStarletteApplication(
74+
agent_card=public_agent_card,
75+
http_handler=request_handler,
76+
).build()
77+
78+
Args:
79+
agent: The AI agent to execute.
80+
stream: Whether to stream the agent response. Defaults to False.
81+
run_kwargs: Additional keyword arguments to pass to the agent's run method.
82+
"""
83+
84+
def __init__(self, agent: SupportsAgentRun, stream: bool = False, run_kwargs: Mapping[str, Any] | None = None):
85+
"""Initialize the A2AExecutor with the specified agent.
86+
87+
Args:
88+
agent: The AI agent or workflow to execute.
89+
stream: Whether to stream the agent response. Defaults to False.
90+
run_kwargs: Additional keyword arguments to pass to the agent's run method.
91+
Cannot contain 'session' or 'stream' as these are managed by the executor.
92+
93+
Raises:
94+
ValueError: If run_kwargs contains 'session' or 'stream'.
95+
"""
96+
super().__init__()
97+
self._agent: SupportsAgentRun = agent
98+
self._stream: bool = stream
99+
if run_kwargs:
100+
if "session" in run_kwargs:
101+
raise ValueError("run_kwargs cannot contain 'session' as it is managed by the executor.")
102+
if "stream" in run_kwargs:
103+
raise ValueError("run_kwargs cannot contain 'stream' as it is managed by the executor.")
104+
self._run_kwargs: Mapping[str, Any] = run_kwargs or {}
105+
106+
@override
107+
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
108+
"""Cancel agent execution for the given request context.
109+
110+
Uses a TaskUpdater to send a cancellation event through the provided event queue.
111+
112+
Args:
113+
context: The request context identifying the task to cancel.
114+
event_queue: The event queue to publish the cancellation event to.
115+
116+
Raises:
117+
ValueError: If context_id is not provided in the RequestContext.
118+
"""
119+
if context.context_id is None:
120+
raise ValueError("Context ID must be provided in the RequestContext")
121+
122+
updater = TaskUpdater(
123+
event_queue=event_queue,
124+
task_id=context.task_id or "",
125+
context_id=context.context_id,
126+
)
127+
128+
await updater.cancel()
129+
130+
@override
131+
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
132+
"""Execute the agent with the given context and event queue.
133+
134+
Orchestrates the agent execution process: sets up the agent session,
135+
executes the agent, processes response messages, and handles errors with appropriate task status updates.
136+
"""
137+
if context.context_id is None:
138+
raise ValueError("Context ID must be provided in the RequestContext")
139+
if context.message is None:
140+
raise ValueError("Message must be provided in the RequestContext")
141+
142+
query = context.get_user_input()
143+
task = context.current_task
144+
145+
if not task:
146+
task = new_task(context.message)
147+
await event_queue.enqueue_event(task)
148+
149+
updater = TaskUpdater(event_queue, task.id, context.context_id)
150+
await updater.submit()
151+
152+
try:
153+
await updater.start_work()
154+
155+
session = self._agent.create_session(session_id=task.context_id)
156+
157+
if self._stream:
158+
await self._run_stream(query, session, updater)
159+
else:
160+
await self._run(query, session, updater)
161+
162+
# Mark as complete
163+
await updater.complete()
164+
except CancelledError:
165+
await updater.update_status(state=TaskState.canceled, final=True)
166+
except Exception as e:
167+
logger.exception("A2AExecutor encountered an error during execution.", exc_info=e)
168+
await updater.update_status(
169+
state=TaskState.failed,
170+
final=True,
171+
message=updater.new_agent_message([Part(root=TextPart(text=str(e)))]),
172+
)
173+
174+
async def _run_stream(self, query: Any, session: AgentSession, updater: TaskUpdater) -> None:
175+
"""Run the agent in streaming mode and publish updates to the task updater."""
176+
response_stream = self._agent.run(query, session=session, stream=True, **self._run_kwargs)
177+
streamed_artifact_ids: set[str] = set()
178+
await (
179+
response_stream.with_transform_hook(
180+
partial(self.handle_events, updater=updater, streamed_artifact_ids=streamed_artifact_ids)
181+
)
182+
).get_final_response()
183+
184+
async def _run(self, query: Any, session: AgentSession, updater: TaskUpdater) -> None:
185+
"""Run the agent in non-streaming mode and publish messages to the task updater."""
186+
response = await self._agent.run(query, session=session, stream=False, **self._run_kwargs)
187+
response_messages = response.messages
188+
189+
if not isinstance(response_messages, list):
190+
response_messages = [response_messages]
191+
192+
for message in response_messages:
193+
await self.handle_events(message, updater)
194+
195+
async def handle_events(
196+
self, item: Message | AgentResponseUpdate, updater: TaskUpdater, streamed_artifact_ids: set[str] | None = None
197+
) -> None:
198+
"""Convert agent response items (Messages or Updates) to A2A protocol events.
199+
200+
Processes Message or AgentResponseUpdate objects and converts them into A2A protocol format.
201+
Handles text, data, and URI content. USER role messages are skipped.
202+
203+
Users can override this method in a subclass to implement custom transformations
204+
from their agent's output format to A2A protocol events.
205+
206+
Args:
207+
item: The agent response item (Message or AgentResponseUpdate) to process.
208+
updater: The task updater to publish events to.
209+
streamed_artifact_ids: A set of artifact IDs that have already been streamed.
210+
Used to prevent duplicate updates for the same artifact.
211+
212+
Example:
213+
.. code-block:: python
214+
215+
class CustomA2AExecutor(A2AExecutor):
216+
async def handle_events(
217+
self,
218+
item: Message | AgentResponseUpdate,
219+
updater: TaskUpdater,
220+
streamed_artifact_ids: set[str] | None = None,
221+
) -> None:
222+
# Custom logic to transform item contents
223+
if item.role == "assistant" and item.contents:
224+
parts = [Part(root=TextPart(text=f"Custom: {item.contents[0].text}"))]
225+
await updater.update_status(
226+
state=TaskState.working,
227+
message=updater.new_agent_message(parts=parts),
228+
)
229+
else:
230+
await super().handle_events(item, updater)
231+
"""
232+
role = getattr(item, "role", None)
233+
if role == "user":
234+
# This is a user message, we can ignore it in the context of task updates
235+
return
236+
237+
parts: list[Part] = []
238+
metadata = getattr(item, "additional_properties", None)
239+
240+
# AgentResponseUpdate uses 'contents', Message uses 'contents'
241+
contents = getattr(item, "contents", [])
242+
243+
for content in contents:
244+
if content.type == "text" and content.text:
245+
parts.append(Part(root=TextPart(text=content.text)))
246+
elif content.type == "data" and content.uri:
247+
base64_str = get_uri_data(content.uri)
248+
parts.append(Part(root=FilePart(file=FileWithBytes(bytes=base64_str, mime_type=content.media_type))))
249+
elif content.type == "uri" and content.uri:
250+
parts.append(Part(root=FilePart(file=FileWithUri(uri=content.uri, mime_type=content.media_type))))
251+
else:
252+
# Silently skip unsupported content types
253+
logger.warning("A2AExecutor does not yet support content type: %s. Omitted.", content.type)
254+
255+
if parts:
256+
if isinstance(item, AgentResponseUpdate):
257+
# For streaming updates, we send TaskArtifactUpdateEvent via add_artifact
258+
await updater.add_artifact(
259+
parts=parts,
260+
artifact_id=item.message_id,
261+
metadata=metadata,
262+
append=(
263+
True
264+
if streamed_artifact_ids is not None and item.message_id in (streamed_artifact_ids or set())
265+
else None
266+
),
267+
)
268+
if item.message_id and streamed_artifact_ids is not None:
269+
streamed_artifact_ids.add(item.message_id)
270+
else:
271+
# For final messages, we send TaskStatusUpdateEvent with 'working' state
272+
await updater.update_status(
273+
state=TaskState.working,
274+
message=updater.new_agent_message(parts=parts, metadata=metadata),
275+
)

0 commit comments

Comments
 (0)