[#4442][feat] Add client-side A2A (Agent2Agent) protocol support to Scaffolding#15350
Open
Achyuthan-S wants to merge 2 commits into
Open
[#4442][feat] Add client-side A2A (Agent2Agent) protocol support to Scaffolding#15350Achyuthan-S wants to merge 2 commits into
Achyuthan-S wants to merge 2 commits into
Conversation
…ding Add a client-side A2A contrib mirroring the MCP contrib: A2AWorker delegates to remote A2A agents, A2AController routes the LLM's tool calls to them and summarizes the replies. Includes a runnable example with a reference agent server and offline unit tests. Closes NVIDIA#4442 Signed-off-by: Achyuthan Sivasankar <achyuthan.sivasankar@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds an Agent2Agent (A2A) scaffolding contrib that can discover remote agents, delegate requests to them, and summarize their replies via an existing generation worker, plus offline unit tests and runnable examples.
Changes:
- Introduces
A2AController,A2AWorker,A2AListTask/A2ASendTask, and ana2a-sdkwrapper (A2AAgentConnection+AgentInfo). - Adds offline unit tests using fake A2A connections to avoid requiring
a2a-sdkand network access. - Adds example scripts and documentation to run an orchestrator and a demo remote agent server.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/unittest/scaffolding/test_a2a_worker.py | Adds offline unit + end-to-end tests for A2A worker/controller behavior. |
| tensorrt_llm/scaffolding/contrib/a2a/a2a_worker.py | Implements a worker to list agents, send messages, and shutdown connections. |
| tensorrt_llm/scaffolding/contrib/a2a/a2a_utils.py | Adds a lazy-import wrapper around a2a-sdk + response text extraction. |
| tensorrt_llm/scaffolding/contrib/a2a/a2a_task.py | Defines list/send task dataclasses used by controller and worker. |
| tensorrt_llm/scaffolding/contrib/a2a/a2a_controller.py | Adds an orchestrator controller that tool-calls remote agents via A2A. |
| tensorrt_llm/scaffolding/contrib/a2a/init.py | Exposes the A2A contrib public API. |
| examples/scaffolding/contrib/a2a/weather_agent_server.py | Provides a minimal demo A2A server to test orchestration locally. |
| examples/scaffolding/contrib/a2a/a2a_run.py | Provides a runnable orchestrator example using OpenAI-compatible generation. |
| examples/scaffolding/contrib/a2a/README.md | Documents setup and usage for the A2A example. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+87
to
+89
| def process(self, tasks: List[Task], **kwargs): | ||
| assert len(tasks) == 1, "A2AController handles a single task at a time." | ||
| result_task = tasks[0] |
Comment on lines
+118
to
+125
| send_tasks = [] | ||
| for tool_call in chat_task.tool_calls: | ||
| args = json.loads(tool_call.function.arguments) | ||
| send_tasks.append( | ||
| A2ASendTask.create_a2a_task( | ||
| tool_call.function.name, args.get("message", ""), self.WorkerTag.A2A | ||
| ) | ||
| ) |
Comment on lines
+40
to
+56
| return { | ||
| "type": "function", | ||
| "function": { | ||
| "name": agent.name, | ||
| "description": description, | ||
| "parameters": { | ||
| "type": "object", | ||
| "properties": { | ||
| "message": { | ||
| "type": "string", | ||
| "description": "The natural-language request to send to this agent.", | ||
| } | ||
| }, | ||
| "required": ["message"], | ||
| }, | ||
| }, | ||
| } |
Comment on lines
+42
to
+48
| async def init_with_urls(cls, urls: List[str]) -> "A2AWorker": | ||
| connections = [] | ||
| for url in urls: | ||
| connection = A2AAgentConnection() | ||
| await connection.connect(url) | ||
| connections.append(connection) | ||
| return cls(connections) |
Comment on lines
+109
to
+124
| async def send_message(self, message: str) -> str: | ||
| """Send a text message to the remote agent and return its text reply.""" | ||
| from a2a.types import MessageSendParams, SendMessageRequest | ||
|
|
||
| request = SendMessageRequest( | ||
| id=str(uuid.uuid4()), | ||
| params=MessageSendParams( | ||
| message={ | ||
| "role": "user", | ||
| "parts": [{"kind": "text", "text": message}], | ||
| "messageId": uuid.uuid4().hex, | ||
| } | ||
| ), | ||
| ) | ||
| response = await self._client.send_message(request) | ||
| return _extract_text_from_response(response) |
Comment on lines
+92
to
+95
| self._httpx_client = httpx.AsyncClient() | ||
| resolver = A2ACardResolver(httpx_client=self._httpx_client, base_url=base_url.rstrip("/")) | ||
| self._agent_card = await resolver.get_agent_card() | ||
| self._client = A2AClient(httpx_client=self._httpx_client, agent_card=self._agent_card) |
Comment on lines
+42
to
+48
| def _extract_text_from_response(response) -> str: | ||
| """Best-effort extraction of textual content from an A2A send-message response. | ||
|
|
||
| The ``a2a-sdk`` response schema has shifted across versions, so we walk the | ||
| structure defensively and fall back to ``str(response)`` if no text part is | ||
| found rather than raising on an unexpected layout. | ||
| """ |
…d JSON Degrade gracefully instead of aborting the controller when a model emits invalid JSON (or a non-object payload) in tool-call arguments. Addresses review feedback on NVIDIA#15350. Signed-off-by: Achyuthan Sivasankar <achyuthan.sivasankar@gmail.com>
Contributor
📝 WalkthroughWalkthroughAdds a new ChangesA2A Scaffolding Contrib Module
Sequence Diagram(s)sequenceDiagram
participant User
participant a2a_run.py
participant ScaffoldingLlm
participant A2AController
participant OpenaiWorker
participant A2AWorker
participant WeatherAgentServer
User->>a2a_run.py: run with --prompt, --agent_urls, --model
a2a_run.py->>ScaffoldingLlm: generate(prompt)
ScaffoldingLlm->>A2AController: process(result_task)
A2AController->>A2AWorker: A2AListTask (discover agents)
A2AWorker->>WeatherAgentServer: connect() / agent card resolve
WeatherAgentServer-->>A2AWorker: AgentInfo (weather skill)
A2AWorker-->>A2AController: result_agents
A2AController->>OpenaiWorker: ChatTask with weather tool schema
OpenaiWorker-->>A2AController: tool_call(weather_agent, message)
A2AController->>A2AWorker: A2ASendTask(weather_agent, message)
A2AWorker->>WeatherAgentServer: send_message()
WeatherAgentServer-->>A2AWorker: "It is sunny in LA, around 75F."
A2AController->>OpenaiWorker: ChatTask (summarize reply)
OpenaiWorker-->>A2AController: final summary text
A2AController-->>ScaffoldingLlm: result_task.output_str
ScaffoldingLlm-->>a2a_run.py: output text
a2a_run.py-->>User: print result
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tests/unittest/scaffolding/test_a2a_worker.py (1)
148-174:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winCoverage is insufficient for controller resilience branches.
tests/unittest/scaffolding/test_a2a_worker.pycurrently validates the main delegation flow, but it does not cover:
- malformed/non-JSON
tool_call.function.argumentshandling intensorrt_llm/scaffolding/contrib/a2a/a2a_controller.py- the no-delegation branch (
finish_reason != "tool_calls") returning direct model outputPlease add focused tests in
tests/unittest/scaffolding/test_a2a_worker.pyfor both paths.Based on learnings from the review instructions for
tests/**: provide actionable test coverage feedback with concrete file targets and sufficiency assessment.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/scaffolding/test_a2a_worker.py` around lines 148 - 174, The test_scaffolding_with_a2a_controller test only covers the happy path for successful delegation. Add two additional test cases to improve coverage: one test that validates handling of malformed or non-JSON tool_call.function.arguments data in the A2AController to ensure it gracefully handles parsing errors, and another test that validates the no-delegation branch where finish_reason is not "tool_calls", confirming that the controller returns direct model output without attempting worker delegation. Both tests should follow the same setup pattern as the existing test but configure mock responses or conditions to trigger these specific code paths in the A2AController.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@examples/scaffolding/contrib/a2a/a2a_run.py`:
- Around line 54-80: The main() function does not guarantee cleanup when
generation fails because the shutdown calls for llm, generation_worker, and
a2a_worker are only executed on the success path. Wrap the generation logic (the
llm.generate_async call and the await future.aresult() call along with the
result printing) in a try block, and move all shutdown calls (llm.shutdown(),
generation_worker.shutdown(), and await a2a_worker.async_shutdown()) into a
finally block to ensure they always execute regardless of whether generation
succeeds or raises an exception.
In `@examples/scaffolding/contrib/a2a/weather_agent_server.py`:
- Around line 51-63: The AgentCard.url is currently using the bind host
parameter directly, which defaults to 0.0.0.0 - a non-routable address that
should not be advertised. Modify the url construction in the build_app function
to normalize 0.0.0.0 to 127.0.0.1 (or another reachable address) for the
advertised endpoint, keeping the bind host separate from the advertised host.
This ensures the AgentCard advertises a reachable URL that clients can actually
connect to.
In `@tensorrt_llm/scaffolding/contrib/a2a/a2a_controller.py`:
- Around line 88-89: Replace the assert statement checking that tasks length
equals 1 with an explicit runtime validation using an if condition that raises a
concrete exception (such as ValueError) when the condition is not met. This
ensures the validation cannot be disabled and will properly catch invalid
controller input at runtime, preventing downstream index or logic failures when
accessing tasks[0].
- Around line 119-124: The json.loads call in the tool-call processing loop can
raise JSONDecodeError and crash the entire request if the model outputs
malformed JSON. Wrap the json.loads(tool_call.function.arguments) call in a
try-except block to catch JSONDecodeError, validate that the parsed args is a
dictionary before attempting to call args.get("message", ""), and skip the
current iteration (continue) if either validation fails. Optionally log a
warning when skipping invalid tool calls to aid debugging of model output
issues.
In `@tensorrt_llm/scaffolding/contrib/a2a/a2a_worker.py`:
- Around line 42-48: The init_with_urls() classmethod has a resource leak: if
any connection fails during the loop, the previously established connections
remain open. Wrap the connection loop in a try-except block to catch exceptions
during the await connection.connect(url) call. When an exception is caught,
iterate through the connections list and call appropriate cleanup (such as
async_shutdown()) on each successfully connected connection before re-raising
the exception. This ensures all resources are properly released when startup
fails partway through.
---
Outside diff comments:
In `@tests/unittest/scaffolding/test_a2a_worker.py`:
- Around line 148-174: The test_scaffolding_with_a2a_controller test only covers
the happy path for successful delegation. Add two additional test cases to
improve coverage: one test that validates handling of malformed or non-JSON
tool_call.function.arguments data in the A2AController to ensure it gracefully
handles parsing errors, and another test that validates the no-delegation branch
where finish_reason is not "tool_calls", confirming that the controller returns
direct model output without attempting worker delegation. Both tests should
follow the same setup pattern as the existing test but configure mock responses
or conditions to trigger these specific code paths in the A2AController.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 10017755-74ac-481d-9ae1-515f5aae423e
📒 Files selected for processing (9)
examples/scaffolding/contrib/a2a/README.mdexamples/scaffolding/contrib/a2a/a2a_run.pyexamples/scaffolding/contrib/a2a/weather_agent_server.pytensorrt_llm/scaffolding/contrib/a2a/__init__.pytensorrt_llm/scaffolding/contrib/a2a/a2a_controller.pytensorrt_llm/scaffolding/contrib/a2a/a2a_task.pytensorrt_llm/scaffolding/contrib/a2a/a2a_utils.pytensorrt_llm/scaffolding/contrib/a2a/a2a_worker.pytests/unittest/scaffolding/test_a2a_worker.py
Comment on lines
+54
to
+80
| async def main(): | ||
| args = parse_arguments() | ||
|
|
||
| client = AsyncOpenAI(api_key=args.API_KEY, base_url=args.base_url) | ||
| generation_worker = OpenaiWorker(client, args.model) | ||
| generation_worker.register_task_handler(ChatTask, chat_handler) | ||
|
|
||
| a2a_worker = await A2AWorker.init_with_urls(args.agent_urls) | ||
|
|
||
| controller = A2AController() | ||
| llm = ScaffoldingLlm( | ||
| controller, | ||
| { | ||
| A2AController.WorkerTag.GENERATION: generation_worker, | ||
| A2AController.WorkerTag.A2A: a2a_worker, | ||
| }, | ||
| ) | ||
|
|
||
| future = llm.generate_async(args.prompt) | ||
| result = await future.aresult() | ||
| print(f"\nresult is {result.outputs[0].text}\n") | ||
|
|
||
| print("shutting down...") | ||
| llm.shutdown() | ||
| generation_worker.shutdown() | ||
| await a2a_worker.async_shutdown() | ||
| print("shut down done") |
Contributor
There was a problem hiding this comment.
Ensure worker shutdown runs even when generation fails.
Cleanup is currently best-effort on success only. Wrap orchestration in try/finally so llm, generation_worker, and a2a_worker are always closed.
Suggested patch
async def main():
args = parse_arguments()
client = AsyncOpenAI(api_key=args.API_KEY, base_url=args.base_url)
generation_worker = OpenaiWorker(client, args.model)
generation_worker.register_task_handler(ChatTask, chat_handler)
a2a_worker = await A2AWorker.init_with_urls(args.agent_urls)
controller = A2AController()
llm = ScaffoldingLlm(
controller,
{
A2AController.WorkerTag.GENERATION: generation_worker,
A2AController.WorkerTag.A2A: a2a_worker,
},
)
-
- future = llm.generate_async(args.prompt)
- result = await future.aresult()
- print(f"\nresult is {result.outputs[0].text}\n")
-
- print("shutting down...")
- llm.shutdown()
- generation_worker.shutdown()
- await a2a_worker.async_shutdown()
- print("shut down done")
+ try:
+ future = llm.generate_async(args.prompt)
+ result = await future.aresult()
+ print(f"\nresult is {result.outputs[0].text}\n")
+ finally:
+ print("shutting down...")
+ llm.shutdown()
+ generation_worker.shutdown()
+ await a2a_worker.async_shutdown()
+ print("shut down done")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@examples/scaffolding/contrib/a2a/a2a_run.py` around lines 54 - 80, The main()
function does not guarantee cleanup when generation fails because the shutdown
calls for llm, generation_worker, and a2a_worker are only executed on the
success path. Wrap the generation logic (the llm.generate_async call and the
await future.aresult() call along with the result printing) in a try block, and
move all shutdown calls (llm.shutdown(), generation_worker.shutdown(), and await
a2a_worker.async_shutdown()) into a finally block to ensure they always execute
regardless of whether generation succeeds or raises an exception.
Comment on lines
+51
to
+63
| def build_app(host: str, port: int) -> A2AStarletteApplication: | ||
| skill = AgentSkill( | ||
| id="weather", | ||
| name="weather", | ||
| description="Returns the current weather for a location.", | ||
| tags=["weather"], | ||
| examples=["What is the weather in LA?"], | ||
| ) | ||
| agent_card = AgentCard( | ||
| name="weather_agent", | ||
| description="A demo agent that reports the weather.", | ||
| url=f"http://{host}:{port}/", | ||
| version="1.0.0", |
Contributor
There was a problem hiding this comment.
Advertise a reachable endpoint, not the bind-all address.
AgentCard.url currently mirrors --host; with the default 0.0.0.0, the card can advertise a non-routable/ambiguous URL. Keep bind host and advertised host separate (or normalize 0.0.0.0 to 127.0.0.1 for local demos), and align a2a_run.py/README defaults accordingly.
Suggested patch
-def build_app(host: str, port: int) -> A2AStarletteApplication:
+def build_app(bind_host: str, port: int, public_host: str | None = None) -> A2AStarletteApplication:
+ advertised_host = public_host or ("127.0.0.1" if bind_host == "0.0.0.0" else bind_host)
skill = AgentSkill(
@@
agent_card = AgentCard(
@@
- url=f"http://{host}:{port}/",
+ url=f"http://{advertised_host}:{port}/",
@@
def main():
@@
- parser.add_argument("--host", default="0.0.0.0")
+ parser.add_argument("--host", default="0.0.0.0")
+ parser.add_argument("--public_host", default=None)
@@
- app = build_app(args.host, args.port)
+ app = build_app(args.host, args.port, args.public_host)🧰 Tools
🪛 ast-grep (0.43.0)
[warning] 61-61: Do not make http calls without encryption
Context: f"http://{host}:{port}/"
Note: [CWE-319].
(requests-http)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@examples/scaffolding/contrib/a2a/weather_agent_server.py` around lines 51 -
63, The AgentCard.url is currently using the bind host parameter directly, which
defaults to 0.0.0.0 - a non-routable address that should not be advertised.
Modify the url construction in the build_app function to normalize 0.0.0.0 to
127.0.0.1 (or another reachable address) for the advertised endpoint, keeping
the bind host separate from the advertised host. This ensures the AgentCard
advertises a reachable URL that clients can actually connect to.
Comment on lines
+88
to
+89
| assert len(tasks) == 1, "A2AController handles a single task at a time." | ||
| result_task = tasks[0] |
Contributor
There was a problem hiding this comment.
Use explicit runtime validation instead of assert for controller input shape.
assert can be disabled, which can turn this into downstream index/logic failures. Raise a concrete exception for non-single-task inputs.
Suggested fix
- assert len(tasks) == 1, "A2AController handles a single task at a time."
+ if len(tasks) != 1:
+ raise ValueError("A2AController handles a single task at a time.")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/scaffolding/contrib/a2a/a2a_controller.py` around lines 88 - 89,
Replace the assert statement checking that tasks length equals 1 with an
explicit runtime validation using an if condition that raises a concrete
exception (such as ValueError) when the condition is not met. This ensures the
validation cannot be disabled and will properly catch invalid controller input
at runtime, preventing downstream index or logic failures when accessing
tasks[0].
Comment on lines
+119
to
+124
| for tool_call in chat_task.tool_calls: | ||
| args = json.loads(tool_call.function.arguments) | ||
| send_tasks.append( | ||
| A2ASendTask.create_a2a_task( | ||
| tool_call.function.name, args.get("message", ""), self.WorkerTag.A2A | ||
| ) |
Contributor
There was a problem hiding this comment.
Harden tool-call argument parsing against malformed model output.
json.loads(tool_call.function.arguments) can raise and abort the whole request. Handle JSONDecodeError and non-object payloads, then skip invalid calls.
Suggested fix
send_tasks = []
for tool_call in chat_task.tool_calls:
- args = json.loads(tool_call.function.arguments)
+ try:
+ args = json.loads(tool_call.function.arguments or "{}")
+ except json.JSONDecodeError:
+ continue
+ if not isinstance(args, dict):
+ continue
+ message = args.get("message", "")
+ if not isinstance(message, str):
+ continue
send_tasks.append(
A2ASendTask.create_a2a_task(
- tool_call.function.name, args.get("message", ""), self.WorkerTag.A2A
+ tool_call.function.name, message, self.WorkerTag.A2A
)
)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/scaffolding/contrib/a2a/a2a_controller.py` around lines 119 -
124, The json.loads call in the tool-call processing loop can raise
JSONDecodeError and crash the entire request if the model outputs malformed
JSON. Wrap the json.loads(tool_call.function.arguments) call in a try-except
block to catch JSONDecodeError, validate that the parsed args is a dictionary
before attempting to call args.get("message", ""), and skip the current
iteration (continue) if either validation fails. Optionally log a warning when
skipping invalid tool calls to aid debugging of model output issues.
Comment on lines
+42
to
+48
| async def init_with_urls(cls, urls: List[str]) -> "A2AWorker": | ||
| connections = [] | ||
| for url in urls: | ||
| connection = A2AAgentConnection() | ||
| await connection.connect(url) | ||
| connections.append(connection) | ||
| return cls(connections) |
Contributor
There was a problem hiding this comment.
Rollback successful connections when one URL fails during startup.
If one connect(url) fails, earlier connections remain open because worker construction aborts before async_shutdown() can run. Add rollback cleanup in init_with_urls().
Suggested fix
`@classmethod`
async def init_with_urls(cls, urls: List[str]) -> "A2AWorker":
connections = []
- for url in urls:
- connection = A2AAgentConnection()
- await connection.connect(url)
- connections.append(connection)
+ try:
+ for url in urls:
+ connection = A2AAgentConnection()
+ try:
+ await connection.connect(url)
+ except Exception:
+ await connection.cleanup()
+ raise
+ connections.append(connection)
+ except Exception:
+ for connection in connections:
+ await connection.cleanup()
+ raise
return cls(connections)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/scaffolding/contrib/a2a/a2a_worker.py` around lines 42 - 48, The
init_with_urls() classmethod has a resource leak: if any connection fails during
the loop, the previously established connections remain open. Wrap the
connection loop in a try-except block to catch exceptions during the await
connection.connect(url) call. When an exception is caught, iterate through the
connections list and call appropriate cleanup (such as async_shutdown()) on each
successfully connected connection before re-raising the exception. This ensures
all resources are properly released when startup fails partway through.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description:
What this does
Follow-up to the MCP work (#4410). Issue #4442 asked for A2A protocol support in
Scaffolding now that MCP tool-calling is in place, plus an example to prove it
works end to end. This adds a client-side A2A integration under
scaffolding/contrib/a2a/, built the same way as the MCP contrib so it shouldfeel familiar to anyone who's looked at that code.
The idea: the generation model sees each reachable remote agent as a callable
tool, decides whether to delegate, and the
A2AWorkerforwards the message overA2A and hands the reply back to the model to fold into a final answer.
This is intentionally the client side only (Scaffolding consuming other agents).
Exposing a Scaffolding pipeline as an A2A server is a natural next step but I
kept it out of scope here to keep the PR focused — happy to do that as a
follow-up if there's interest.
What's included
A2AWorker+A2ASendTask/A2AListTask— discover remote agents and sendthem messages (mirrors
MCPWorker/MCPCallTask/MCPListTask).A2AController— discovers agents, lets the LLM pick one, dispatches the call,and summarizes the result. Same
process()shape asMCPController.A2AAgentConnection— thin wrapper overa2a-sdk. The SDK is imported lazilyso importing the package (and the unit tests) don't need it installed.
examples/scaffolding/contrib/a2a/with a runnableorchestrator and a small reference agent server, plus a README.
tests/unittest/scaffolding/test_a2a_worker.pythatinject fake connections, so they need neither
a2a-sdknor network access.A couple of decisions worth calling out
I reuse the MCP contrib's
ChatTask/chat_handler(imported from thesubmodules directly) rather than duplicating them.
contrib/README.mdexplicitly allows cross-project reuse, and the MCP chat handler is the one that
works against generic OpenAI-compatible endpoints. Easy to inline if you'd
prefer the package fully self-contained.
I did not add
a2a-sdktorequirements.txt— it's an optional dependencypulled in lazily with a clear install message, so it doesn't bloat the base
install.
mcpis in requirements today, so if you'd rather have parity I canadd it.
pytest tests/unittest/scaffolding/test_a2a_worker.py(offline, runs in CI — noa2a-sdkor network required)Manual smoke test (local, needs
a2a-sdk+ an OpenAI-compatible endpoint): runweather_agent_server.py, thena2a_run.pyagainst itNote:
a2a-sdk's server/client API has shifted across versions; the examplescripts target the SDK's published "helloworld" pattern.
cc @WeiHaocheng — please confirm scope / assign if this looks right.
cc @chang-l @Shixiaowei02
Addresses #4442.
Summary by CodeRabbit
New Features
Documentation
Tests