Add core agent A2A adapter - #1456
Conversation
|
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces an A2A (Agent-to-Agent) protocol adapter for the core agent service, including Pydantic schemas, discovery and message-sending endpoints, and comprehensive unit tests. The review feedback focuses on improving robustness and error handling: specifically, wrapping SSE payload JSON parsing in a try-except block to prevent crashes, safely parsing the max_loop_count integer and providing a fallback UUID for uid, and breaking early from the SSE stream processing loop upon encountering an error.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def _parse_sse_payload(chunk: str) -> dict[str, Any] | None: | ||
| for line in chunk.splitlines(): | ||
| if not line.startswith("data:"): | ||
| continue | ||
| payload = line.removeprefix("data:").strip() | ||
| if not payload or payload == "[DONE]": | ||
| return None | ||
| return json.loads(payload) | ||
| return None |
There was a problem hiding this comment.
Parsing the SSE payload with json.loads without error handling can cause the request to crash with a 500 Internal Server Error if the stream contains malformed JSON or non-JSON data. Wrap the parsing logic in a try-except block to handle json.JSONDecodeError gracefully.
| def _parse_sse_payload(chunk: str) -> dict[str, Any] | None: | |
| for line in chunk.splitlines(): | |
| if not line.startswith("data:"): | |
| continue | |
| payload = line.removeprefix("data:").strip() | |
| if not payload or payload == "[DONE]": | |
| return None | |
| return json.loads(payload) | |
| return None | |
| def _parse_sse_payload(chunk: str) -> dict[str, Any] | None: | |
| for line in chunk.splitlines(): | |
| if not line.startswith("data:"): | |
| continue | |
| payload = line.removeprefix("data:").strip() | |
| if not payload or payload == "[DONE]": | |
| return None | |
| try: | |
| return json.loads(payload) | |
| except json.JSONDecodeError: | |
| return None | |
| return None |
There was a problem hiding this comment.
Fixed in f901f33: _parse_sse_payload now catches malformed JSON and ignores non-dict payloads instead of letting the request fail. Added regression coverage in test_collect_completion_text_ignores_malformed_sse_payload.
There was a problem hiding this comment.
Follow-up after branch rewrite: the malformed SSE JSON fix is now in current public head 178eab5. _parse_sse_payload catches JSON decode errors and ignores non-dict payloads; regression coverage is test_collect_completion_text_ignores_malformed_sse_payload. Verified locally with PYTHONPATH=.. uv run --python 3.12 python -m pytest tests/test_a2a.py -q -> 8 passed.
| max_loop_count = metadata.get("max_loop_count", os.getenv("A2A_MAX_LOOP_COUNT", 5)) | ||
|
|
||
| return CustomCompletionInputs( | ||
| uid=( | ||
| _metadata_string(metadata, "uid") | ||
| or _metadata_string(request.message.metadata, "uid") | ||
| or request.message.context_id | ||
| or request.message.message_id | ||
| )[:64], | ||
| messages=[LLMMessage(role="user", content=text)], | ||
| stream=False, | ||
| meta_data={ | ||
| "caller": "a2a_http_json", | ||
| "caller_sid": request.message.message_id, | ||
| "workflow_id": _metadata_string(metadata, "workflow_id"), | ||
| "run_id": request.message.task_id, | ||
| "node_id": _metadata_string(metadata, "node_id"), | ||
| }, | ||
| model_config=model_config, | ||
| instruction=_metadata_mapping(metadata, "instruction"), | ||
| plugin=_metadata_mapping(metadata, "plugin"), | ||
| max_loop_count=int(max_loop_count), | ||
| ) |
There was a problem hiding this comment.
Parsing max_loop_count directly using int() without validation can raise a ValueError or TypeError if the environment variable or metadata contains an invalid integer representation. Additionally, if uid is empty, it can cause tracing issues with Span. Robustly parse max_loop_count with a fallback and ensure uid has a fallback UUID. When using os.getenv, default values should be provided as strings to prevent type issues.
try:
max_loop_count = int(metadata.get("max_loop_count") or os.getenv("A2A_MAX_LOOP_COUNT", "5"))
except (ValueError, TypeError):
max_loop_count = 5
return CustomCompletionInputs(
uid=(
_metadata_string(metadata, "uid")
or _metadata_string(request.message.metadata, "uid")
or request.message.context_id
or request.message.message_id
or str(uuid.uuid4())
)[:64],
messages=[LLMMessage(role="user", content=text)],
stream=False,
meta_data={
"caller": "a2a_http_json",
"caller_sid": request.message.message_id,
"workflow_id": _metadata_string(metadata, "workflow_id"),
"run_id": request.message.task_id,
"node_id": _metadata_string(metadata, "node_id"),
},
model_config=model_config,
instruction=_metadata_mapping(metadata, "instruction"),
plugin=_metadata_mapping(metadata, "plugin"),
max_loop_count=max_loop_count,
)References
- When using os.getenv in Python, default values should be provided as strings, not integers, to prevent TypeError if the environment variable is not set.
There was a problem hiding this comment.
Fixed in f901f33: max loop count parsing now falls back safely when metadata/env values are invalid, and blank request identifiers now get a UUID fallback. Added regression coverage for invalid loop counts and missing IDs.
There was a problem hiding this comment.
Follow-up after branch rewrite: the invalid max loop count and blank request ID fixes are now in current public head 178eab5. Invalid metadata/env loop counts fall back safely, and blank request identifiers get a UUID fallback; regression coverage covers invalid loop count and missing IDs. Verified locally with PYTHONPATH=.. uv run --python 3.12 python -m pytest tests/test_a2a.py -q -> 8 passed.
| if payload.get("code", 0) != 0: | ||
| error_message = str(payload.get("message") or "A2A agent execution failed") |
There was a problem hiding this comment.
When an error code is received in the SSE stream, the loop should break early to prevent unnecessary processing of subsequent chunks and potential secondary errors.
| if payload.get("code", 0) != 0: | |
| error_message = str(payload.get("message") or "A2A agent execution failed") | |
| if payload.get("code", 0) != 0: | |
| error_message = str(payload.get("message") or "A2A agent execution failed") | |
| break |
There was a problem hiding this comment.
Fixed in f901f33: stream collection now breaks immediately after an error payload, so later chunks are not processed after the error. Added regression coverage in test_collect_completion_text_stops_after_error_code.
There was a problem hiding this comment.
Follow-up after branch rewrite: the stream error handling fix is now in current public head 178eab5. Stream collection breaks immediately after an error payload so later chunks are not processed; regression coverage is test_collect_completion_text_stops_after_error_code. Verified locally with PYTHONPATH=.. uv run --python 3.12 python -m pytest tests/test_a2a.py -q -> 8 passed.
0f6143c to
fcce485
Compare
Signed-off-by: Rajesh Digambar Bagul <102693488+Rajesh270712@users.noreply.github.com>
Signed-off-by: Rajesh Digambar Bagul <102693488+Rajesh270712@users.noreply.github.com>
f901f33 to
178eab5
Compare
dongjiang1989
left a comment
dongjiang1989
left a comment
There was a problem hiding this comment.
Good job. @Rajesh270712
We need tasks in runtime and add e2e test case
| organization="iFLYTEK", | ||
| url="https://github.com/iflytek/astron-agent", | ||
| ), | ||
| version="0.1.0", |
There was a problem hiding this comment.
(nit) We need keep version aligned with release version.
| """Build public A2A discovery metadata for the core agent service.""" | ||
|
|
||
| interface_url = f"{_public_base_url()}/agent/v1/a2a" | ||
| return A2AAgentCard( |
There was a problem hiding this comment.
Miss authentication struct. Authentication requirements for the agent.
| @@ -0,0 +1,342 @@ | |||
| """A2A protocol adapter for the core agent service.""" | |||
There was a problem hiding this comment.
Miss task completion. About: Task, TaskStatus, TaskStatusUpdateEvent, TaskArtifactUpdateEvent, TaskSendParams etc.
Expose task-oriented A2A send/get/events routes backed by the core agent completion runtime. Record completed task state and status/artifact events so A2A clients can retrieve runtime task output after execution. Signed-off-by: Rajesh Digambar Bagul <102693488+Rajesh270712@users.noreply.github.com>
|
Thanks for the review. I pushed Changes in that head:
Validation:
Note: the runtime task store is in-process memory for this adapter follow-up; I did not add persistence or cross-worker synchronization. |
Signed-off-by: Rajesh Digambar Bagul <102693488+Rajesh270712@users.noreply.github.com>
|
I pushed The change declares Validation:
Public CI note: GitHub currently reports the new CodeQL/CI workflow runs for |
|
Thanks a lot for your great contribution! This pull request contains too many changes in one batch. Could you please split it into 3 separate smaller PRs as outlined below for easier review and incremental merge:
Let me know once you’ve split them out, I’ll start reviewing each one promptly. |
|
Thanks for the guidance. I split the work into the requested smaller pieces:
Part 2 and 3 are stacked review PRs in my fork so their diffs stay small. GitHub does not let me open an upstream PR against a base branch that only exists in my fork, so after #1480 lands I can retarget/open part 2 upstream, then do the same for part 3 after part 2. Validation run locally with Python 3.12:
|
|
Closing in favor of the split stack starting with #1480 — we'll review the pieces there. Thanks for splitting it up. |
4 participants
Summary
Adds a focused A2A adapter for the core agent service, including discovery, message sending, runtime task endpoints, and e2e-style task coverage.
Related Issue
Fixes #709.
Problem
Issue #709 asks for Astron Agent to expose an A2A-compatible core agent surface so external A2A clients can discover the agent and send messages/tasks. The existing service did not expose an A2A discovery card,
message:sendHTTP adapter, or task runtime endpoints forcore/agent.Change
/.well-known/agent-card.jsonand/agent/v1/a2a/agent-card.jsonwith HTTP+JSON interface metadata, capabilities, skills, text I/O modes, and the existingx-consumer-usernameheader auth shape./agent/v1/a2a/message:sendto map A2A text parts into the existingCustomChatCompletionrunner and return A2A task envelopes for completed, submitted, and failed states./agent/v1/a2a:POST /tasks:send,GET /tasks/{task_id}, andGET /tasks/{task_id}/events.TypeAliasso the repo mypy quality check accepts the runtime task event annotations.TestCliente2e-style send/get/events task flow.Tests
Latest type-only CI fix validation:
cd core/agent && .venv/bin/python -m mypy --disallow-untyped-defs --disallow-incomplete-defs --check-untyped-defs --no-implicit-optional --ignore-missing-imports --explicit-package-bases .-> Success, no issues in 67 source filescd core/agent && PYTHONPATH=.. .venv/bin/python -m pytest tests/test_a2a.py -q-> 9 passed, 3 warningsflake8on the touched file -> passedisort --check-only --profile blackon the touched file -> passedblack --checkon the touched file -> passedcompileall -q api/v1/a2a.py-> passedgit diff --check HEAD~1..HEADandgit diff --check-> passedRuntime task/e2e follow-up validation before the type-only CI fix:
PYTHONPATH=.. .venv/bin/python -m pytest tests/test_a2a.py -q-> 9 passed, 3 warningsPYTHONPATH=.. .venv/bin/python -m pytest tests/test_router_and_schemas.py tests/test_main.py tests/test_a2a.py -q-> 22 passed, 5 warningsPYTHONPATH=.. .venv/bin/python -m pytest tests -q-> 183 passed, 24 warnings.venv/bin/python -m black --check api/v1/a2a.py api/schemas/a2a.py tests/test_a2a.py-> passed.venv/bin/python -m isort --check-only --profile black api/v1/a2a.py api/schemas/a2a.py tests/test_a2a.py-> passed.venv/bin/python -m flake8 --extend-ignore=E501 api/v1/a2a.py api/schemas/a2a.py tests/test_a2a.py-> passedPYTHONPATH=.. .venv/bin/python -m compileall -q api/v1/a2a.py api/schemas/a2a.py tests/test_a2a.py-> passedRisk Notes
0e0110f19786c0bb22c8afae03e21f377a8214e8currently showaction_required, so their jobs have not run yet.A2A_MODEL_*environment defaults.Maintainer Attention
Please confirm whether the endpoint placement, auth declaration, A2A schema shape, and in-process task runtime approach match the direction you want for
core/agent.