feat(a2a): expose CUGA over A2A v0.3 + working two-CUGA demo#323
Open
laredo wants to merge 3 commits into
Open
feat(a2a): expose CUGA over A2A v0.3 + working two-CUGA demo#323laredo wants to merge 3 commits into
laredo wants to merge 3 commits into
Conversation
added 3 commits
June 9, 2026 15:34
Adds a self-contained A2A server module mounted only when settings.a2a.enabled is true, so disabled deployments are byte-identical to before. Exposes /.well-known/agent[-card].json and a JSON-RPC /a2a endpoint supporting message/send and SSE message/stream, with a lifecycle adapter that maps CUGA graph events onto A2A TaskStatusUpdateEvents (HITL -> input_required, final -> completed, always emits a terminal). Production graph wiring is left as a follow-up; main.py mounts a placeholder runner that returns a clean working->completed lifecycle.
Wires CUGA's outbound and inbound A2A halves end-to-end against the installed a2a-sdk 1.0.2. Replaces the placeholder A2A runner with a lazily-instantiated CugaSupervisor that routes inbound message/send through the real graph; rewrites delegate_task_via_a2a_sdk to speak v0.3 JSON-RPC directly (the SDK 1.x removed A2AClient and MessageSendParams that the prior implementation imported, so HTTP-A2A delegation was silently no-op). Adds docs/examples/a2a_two_cuga/ — provider.yaml, consumer.yaml, two launch scripts, README, and a preflight that loads .env and patches the saved cuga.db agent_configs row so MODEL_NAME survives _apply_published_config (see issue tracking the underlying bug). Tests: 38 passed, 1 xfailed (auth-token forwarding, intentionally deferred). New test_outbound_protocol.py covers _extract_text_from_task, _agent_card_rpc_url, end-to-end roundtrip via httpx.ASGITransport, and JSON-RPC error envelope handling.
The a2a-sdk 1.x A2ACardResolver.get_agent_card() returns a protobuf
AgentCard whose URL lives at supported_interfaces[0].url with a
protocol_binding ("JSONRPC"/"GRPC"/...) — not at the top-level .url
attribute that the v0.3 pydantic shape uses. The previous
_agent_card_rpc_url() only looked at .url, so every cross-process
delegation through delegate_task_via_a2a_sdk() failed with
"Agent card carries no URL" once the supervisor flow handed in the
real (proto) card from the resolver.
Now we try .url first (pydantic / direct callers), then walk
.supported_interfaces preferring an entry whose protocol_binding is
JSONRPC, falling back to fallback_base. Verified end-to-end against
the two-CUGA demo: provider on :8002 receives POST /a2a → 200 with
real digital_sales output round-tripping back to the consumer.
Tests: +2 cases covering the proto shape and JSONRPC-vs-GRPC
preference. 40 passed, 1 xfailed.
Contributor
📝 WalkthroughWalkthroughThis PR implements bidirectional A2A v0.3 protocol support for CUGA. It introduces an inbound FastAPI router advertising agents via well-known endpoints and JSON-RPC, rewrites the outbound client for direct HTTP JSON-RPC, adds configuration toggles, and provides a complete two-CUGA example with provider and consumer setup. ChangesA2A v0.3 Protocol Support
Sequence DiagramsequenceDiagram
participant ClientCUGA as Client CUGA<br/>(Delegation)
participant ServerRouter as Server Router<br/>(/a2a endpoint)
participant SupervisorRunner as Supervisor<br/>(via runner)
participant ResponseAdapter as Task Adapter<br/>(event → A2A)
ClientCUGA->>ServerRouter: POST /a2a<br/>message/send JSON-RPC
ServerRouter->>SupervisorRunner: runner.run(message)
SupervisorRunner->>ResponseAdapter: stream events
ResponseAdapter->>ResponseAdapter: convert to TaskStatusUpdateEvent
ResponseAdapter->>ServerRouter: buffer Task with history
ServerRouter->>ClientCUGA: JSON-RPC result{Task}
ClientCUGA->>ClientCUGA: extract_text_from_task()
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested labels
Suggested reviewers
Poem
🚥 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 docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| assert card.name == "cuga" | ||
| assert card.description == "CUGA exposed over A2A." | ||
| assert card.version == "1.2.3" | ||
| assert str(card.url).startswith("https://cuga.example") |
![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: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/cuga/backend/server/a2a/router.py (1)
1-183:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winRuff format check is failing for this file.
CI indicates this file would be reformatted; run formatter before merge.
🤖 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 `@src/cuga/backend/server/a2a/router.py` around lines 1 - 183, Ruff formatting errors were detected in this file; run the repo's formatter and commit the changes so the CI passes. Fix by running the project's formatter (e.g. ruff format or the configured pre-commit hooks / black + ruff) on this file (affecting functions like build_router, _run_and_collect, _sse_stream, and jsonrpc_endpoint), review the modified whitespace/ordering changes, and add/commit the reformatted file before merging.Source: Pipeline failures
🧹 Nitpick comments (1)
tests/unit/a2a/test_agent_card.py (1)
39-39: ⚡ Quick winUse exact URL assertion for stronger contract coverage.
startswith("https://cuga.example")can pass unintended values (e.g., extra path/host variants). Since fixture input is fixed, assert the exact URL string to tighten the test contract.Suggested change
- assert str(card.url).startswith("https://cuga.example") + assert str(card.url) == "https://cuga.example/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 `@tests/unit/a2a/test_agent_card.py` at line 39, Replace the loose startswith assertion with an exact equality check: change the assertion that currently uses str(card.url).startswith("https://cuga.example") to assert the full expected URL string (e.g., assert str(card.url) == "https://cuga.example") in the test where the card fixture is used (reference variable card and its url property in tests/unit/a2a/test_agent_card.py) so the test validates the exact URL value.
🤖 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 `@docs/examples/a2a_two_cuga/_preflight.py`:
- Around line 63-64: Replace the broad except in the JSON parsing block with a
specific except json.JSONDecodeError that logs the offending row id (rowid) and
continues; locate the try/except around the JSON load operation in _preflight.py
(the loop that processes saved configs) and change it to catch
json.JSONDecodeError, call the existing logger or print with a clear message
including rowid, and then continue so malformed rows are visible for debugging.
In `@docs/examples/a2a_two_cuga/README.md`:
- Around line 11-21: The fenced ASCII-art block in README.md is missing a
language tag which triggers MD040; update the opening fence for the block
starting at the diagram (the triple backticks before the diagram) to include a
language identifier (e.g., text) so the block becomes ```text, leaving the
content unchanged and keeping the closing triple backticks as-is.
In `@src/cuga/backend/cuga_graph/nodes/cuga_supervisor/a2a_protocol.py`:
- Around line 228-233: The code currently treats any non-dict or failed peer
result as success; modify the logic around body/result_obj (the variables `body`
and `result_obj`) in the a2a response handler so that if `body` indicates a peer
task failure or `result_obj` is not a dict you return a failure-shaped response
(e.g., {"result": "", "variables": {}, "status": "failure", "error": <message>})
instead of status "success"; when `result_obj` is present but contains an error
state propagate that error (include its message/details) and only call
`_extract_text_from_task(result_obj)` and return status "success" when the peer
state and result payload shape are valid.
- Around line 158-168: The loop that picks an interface URL for
agent_card.supported_interfaces currently remembers the first interface
regardless of protocol_binding, which can select non-HTTP bindings (e.g., GRPC)
and break the HTTP JSON-RPC sender; change the logic in that loop so you only
set the fallback url when the iface_url has an HTTP(S) scheme (e.g.,
iface_url.lower().startswith("http")) or the protocol_binding is
empty/explicitly HTTP, and continue to prefer protocol_binding == "JSONRPC" as
the primary selector (i.e., keep the existing JSONRPC break path, but replace
the unconditional fallback assignment to url with a guarded assignment that only
accepts HTTP(S)-compatible interfaces).
In `@src/cuga/backend/server/a2a/router.py`:
- Around line 127-134: The SSE generator _sse_stream must catch exceptions from
runner.run(...) and stream_events_to_a2a(...) and emit a JSON-RPC error frame
instead of just closing the stream; wrap the run/iteration logic in try/except,
build an error response via _rpc_result(rpc_id, {"error": {...}}) including the
exception message/type (or a sanitized message) and yield it as a JSON string,
then return/stop the generator; apply the same pattern to the other SSE
generator block (the one also using runner.run and stream_events_to_a2a) so any
runtime error produces a single JSON-RPC error frame to clients rather than a
silent broken stream.
- Around line 164-165: The handler currently returns raw exception text via
JSONResponse(_rpc_error(rpc_id, _INTERNAL_ERROR, "Internal error", str(exc))),
which leaks internal details; instead log the full exception server-side (e.g.,
logging.exception or the existing logger) and change the JSON-RPC response to
return a generic message only (omit str(exc)), keeping the same rpc_id and
_INTERNAL_ERROR constants and using _rpc_error to build the response; ensure the
logged entry includes the exception and stack for debugging while the wire
response contains no internal exception data.
In `@src/cuga/backend/server/a2a/task_adapter.py`:
- Around line 98-107: The terminal-event branch in _is_final(...) currently
always emits TaskState.completed even for error terminal events; update the
branch that yields TaskStatusUpdateEvent so it inspects the event (e.g., ev.name
or ev.type via _event_text/ev.name) and chooses TaskState.failed (or the
project’s error state enum) when the final event represents an error, and only
uses TaskState.completed for successful terminals; preserve the existing message
construction via _message(_event_text(ev) ..., f"{task_id}-final", context_id)
and set the TaskStatus.state accordingly in the TaskStatus passed to
TaskStatusUpdateEvent.
In `@src/cuga/backend/server/main.py`:
- Around line 1632-1719: Ruff reported that src/cuga/backend/server/main.py
needs reformatting; run the project's formatter (ruff/black as configured) to
reformat this file (or the repo) and fix style issues around the A2A block
(e.g., the classes _A2AStreamEvent, _SupervisorA2ARunner, _PlaceholderA2ARunner
and the _a2a_settings_dict/_a2a_supervisor_cfg logic), then re-run the linter
and commit the formatted changes so CI passes.
- Around line 1685-1687: The A2A error event is leaking raw exception text via
f"A2A handler error: {exc}"; in the except block that catches Exception (where
logger.exception(...) and yield _A2AStreamEvent(...) are called) change the
yielded event to include a generic error message (e.g. "A2A handler error")
without including exc or any internal details, keep the logger.exception call
as-is so full exception details remain in server logs, and ensure the
_A2AStreamEvent still sets final=True.
In `@tests/integration/a2a/test_outbound_protocol.py`:
- Around line 87-90: Change mutable list class attributes used in the test
doubles to immutable tuples: replace occurrences of supported_interfaces =
[_Iface()] in the _ProtoCard test double (and the other test double at lines
~107-110) with supported_interfaces = (_Iface(),) so the class-level attribute
is immutable and avoids shared mutable state and the RUF012 warning; update any
other similar class-level list literals in those test classes to tuple
equivalents.
---
Outside diff comments:
In `@src/cuga/backend/server/a2a/router.py`:
- Around line 1-183: Ruff formatting errors were detected in this file; run the
repo's formatter and commit the changes so the CI passes. Fix by running the
project's formatter (e.g. ruff format or the configured pre-commit hooks / black
+ ruff) on this file (affecting functions like build_router, _run_and_collect,
_sse_stream, and jsonrpc_endpoint), review the modified whitespace/ordering
changes, and add/commit the reformatted file before merging.
---
Nitpick comments:
In `@tests/unit/a2a/test_agent_card.py`:
- Line 39: Replace the loose startswith assertion with an exact equality check:
change the assertion that currently uses
str(card.url).startswith("https://cuga.example") to assert the full expected URL
string (e.g., assert str(card.url) == "https://cuga.example") in the test where
the card fixture is used (reference variable card and its url property in
tests/unit/a2a/test_agent_card.py) so the test validates the exact URL value.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 48ff2f51-8714-49e3-ace5-73767ee1bcd8
📒 Files selected for processing (21)
docs/examples/a2a_two_cuga/README.mddocs/examples/a2a_two_cuga/_preflight.pydocs/examples/a2a_two_cuga/consumer.supervisor.yamldocs/examples/a2a_two_cuga/provider.supervisor.yamldocs/examples/a2a_two_cuga/run_consumer.shdocs/examples/a2a_two_cuga/run_provider.shsrc/cuga/backend/cuga_graph/nodes/cuga_supervisor/a2a_protocol.pysrc/cuga/backend/server/a2a/__init__.pysrc/cuga/backend/server/a2a/_a2a_types.pysrc/cuga/backend/server/a2a/agent_card.pysrc/cuga/backend/server/a2a/router.pysrc/cuga/backend/server/a2a/task_adapter.pysrc/cuga/backend/server/main.pysrc/cuga/settings.tomltests/integration/a2a/conftest.pytests/integration/a2a/test_a2a_client_roundtrip.pytests/integration/a2a/test_a2a_router.pytests/integration/a2a/test_cuga_to_cuga.pytests/integration/a2a/test_outbound_protocol.pytests/unit/a2a/test_agent_card.pytests/unit/a2a/test_task_adapter.py
Comment on lines
+63
to
+64
| except Exception: | ||
| continue |
Contributor
There was a problem hiding this comment.
Don’t silently swallow JSON parse failures.
At Line 63, except Exception: continue hides malformed saved configs and makes preflight/debugging unnecessarily opaque. Catch json.JSONDecodeError and log rowid so users can fix corrupted rows.
Suggested fix
- except Exception:
- continue
+ except json.JSONDecodeError as exc:
+ print(f"[preflight] skipped rowid={rowid}: invalid JSON ({exc})", file=sys.stderr)
+ continue📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| except Exception: | |
| continue | |
| except json.JSONDecodeError as exc: | |
| print(f"[preflight] skipped rowid={rowid}: invalid JSON ({exc})", file=sys.stderr) | |
| continue |
🧰 Tools
🪛 Ruff (0.15.15)
[error] 63-64: try-except-continue detected, consider logging the exception
(S112)
[warning] 63-63: Do not catch blind exception: Exception
(BLE001)
🤖 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 `@docs/examples/a2a_two_cuga/_preflight.py` around lines 63 - 64, Replace the
broad except in the JSON parsing block with a specific except
json.JSONDecodeError that logs the offending row id (rowid) and continues;
locate the try/except around the JSON load operation in _preflight.py (the loop
that processes saved configs) and change it to catch json.JSONDecodeError, call
the existing logger or print with a clear message including rowid, and then
continue so malformed rows are visible for debugging.
Source: Linters/SAST tools
Comment on lines
+11
to
+21
| ``` | ||
| ┌──────────────────────────┐ ┌──────────────────────────┐ | ||
| user │ CUGA-1 (consumer) │ A2A v0.3 │ CUGA-2 (provider) │ | ||
| ──→ │ http://localhost:7860/ │ ─JSON-RPC──→ │ http://localhost:8002/ │ | ||
| chat │ no local tools │ /a2a │ digital_sales toolset │ | ||
| │ supervisor: 1 ext agent │ │ A2A inbound enabled │ | ||
| └──────────────────────────┘ └──────────────────────────┘ | ||
| ▲ ▲ | ||
| │ shared │ | ||
| └─── http://localhost:8001 (registry) ─────┘ | ||
| ``` |
Contributor
There was a problem hiding this comment.
Add a language tag to the fenced code block.
At Line 11, the fenced block is missing a language identifier, which triggers MD040 and can fail markdown linting.
Suggested fix
-```
+```text
┌──────────────────────────┐ ┌──────────────────────────┐
user │ CUGA-1 (consumer) │ A2A v0.3 │ CUGA-2 (provider) │
──→ │ http://localhost:7860/ │ ─JSON-RPC──→ │ http://localhost:8002/ │
@@
│ shared │
└─── http://localhost:8001 (registry) ─────┘</details>
<!-- suggestion_start -->
<details>
<summary>📝 Committable suggestion</summary>
> ‼️ **IMPORTANT**
> Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
```suggestion
🧰 Tools
🪛 markdownlint-cli2 (0.22.1)
[warning] 11-11: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🤖 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 `@docs/examples/a2a_two_cuga/README.md` around lines 11 - 21, The fenced
ASCII-art block in README.md is missing a language tag which triggers MD040;
update the opening fence for the block starting at the diagram (the triple
backticks before the diagram) to include a language identifier (e.g., text) so
the block becomes ```text, leaving the content unchanged and keeping the closing
triple backticks as-is.
Source: Linters/SAST tools
Comment on lines
+158
to
+168
| for iface in getattr(agent_card, "supported_interfaces", None) or []: | ||
| iface_url = (getattr(iface, "url", None) or "").strip() | ||
| binding = (getattr(iface, "protocol_binding", None) or "").upper() | ||
| if not iface_url: | ||
| continue | ||
| if binding == "JSONRPC": | ||
| url = iface_url | ||
| break | ||
| if not url: | ||
| url = iface_url # remember first interface; keep looking for JSONRPC | ||
|
|
Contributor
There was a problem hiding this comment.
Avoid selecting non-JSONRPC interface URLs for an HTTP JSON-RPC sender.
At Line 166-167, the fallback keeps the first interface URL even when protocol_binding is not JSONRPC (e.g., GRPC). This can produce an invalid POST target for httpx in Line 220.
Suggested fix
- if not url:
- for iface in getattr(agent_card, "supported_interfaces", None) or []:
- iface_url = (getattr(iface, "url", None) or "").strip()
- binding = (getattr(iface, "protocol_binding", None) or "").upper()
- if not iface_url:
- continue
- if binding == "JSONRPC":
- url = iface_url
- break
- if not url:
- url = iface_url # remember first interface; keep looking for JSONRPC
+ if not url:
+ first_http_candidate = ""
+ for iface in getattr(agent_card, "supported_interfaces", None) or []:
+ iface_url = (getattr(iface, "url", None) or "").strip()
+ binding = (getattr(iface, "protocol_binding", None) or "").upper()
+ if not iface_url:
+ continue
+ if binding == "JSONRPC":
+ url = iface_url
+ break
+ if not binding and iface_url.startswith(("http://", "https://")) and not first_http_candidate:
+ first_http_candidate = iface_url
+ if not url:
+ url = first_http_candidate🤖 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 `@src/cuga/backend/cuga_graph/nodes/cuga_supervisor/a2a_protocol.py` around
lines 158 - 168, The loop that picks an interface URL for
agent_card.supported_interfaces currently remembers the first interface
regardless of protocol_binding, which can select non-HTTP bindings (e.g., GRPC)
and break the HTTP JSON-RPC sender; change the logic in that loop so you only
set the fallback url when the iface_url has an HTTP(S) scheme (e.g.,
iface_url.lower().startswith("http")) or the protocol_binding is
empty/explicitly HTTP, and continue to prefer protocol_binding == "JSONRPC" as
the primary selector (i.e., keep the existing JSONRPC break path, but replace
the unconditional fallback assignment to url with a guarded assignment that only
accepts HTTP(S)-compatible interfaces).
Comment on lines
+228
to
+233
| result_obj = body.get("result") if isinstance(body, dict) else None | ||
| if not isinstance(result_obj, dict): | ||
| return {"result": "", "variables": {}, "status": "success"} | ||
| if isinstance(result_obj, Message): | ||
| text = get_message_text(result_obj) | ||
| elif isinstance(result_obj, Task) and result_obj.history: | ||
| texts = [get_message_text(m) for m in result_obj.history if isinstance(m, Message)] | ||
| text = "\n".join(texts) if texts else "" | ||
| else: | ||
| text = str(result_obj) if result_obj else "" | ||
| return {"result": text or "", "variables": {}, "status": "success"} | ||
|
|
||
| text = _extract_text_from_task(result_obj) | ||
| return {"result": text, "variables": {}, "status": "success"} |
Contributor
There was a problem hiding this comment.
Remote task failures are flattened into status="success".
Line 229-233 returns success unconditionally, even when the peer task state indicates failure or when the result payload shape is invalid. This hides upstream errors and breaks caller-side contract handling.
Suggested fix
result_obj = body.get("result") if isinstance(body, dict) else None
if not isinstance(result_obj, dict):
- return {"result": "", "variables": {}, "status": "success"}
+ raise RuntimeError("A2A response is missing a valid `result` task envelope")
text = _extract_text_from_task(result_obj)
- return {"result": text, "variables": {}, "status": "success"}
+ state = str(((result_obj.get("status") or {}).get("state") or "")).lower()
+ normalized_status = "failed" if ("fail" in state or "error" in state) else "success"
+ out_vars = ((result_obj.get("metadata") or {}).get("variables") or {})
+ return {"result": text, "variables": out_vars, "status": normalized_status}🤖 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 `@src/cuga/backend/cuga_graph/nodes/cuga_supervisor/a2a_protocol.py` around
lines 228 - 233, The code currently treats any non-dict or failed peer result as
success; modify the logic around body/result_obj (the variables `body` and
`result_obj`) in the a2a response handler so that if `body` indicates a peer
task failure or `result_obj` is not a dict you return a failure-shaped response
(e.g., {"result": "", "variables": {}, "status": "failure", "error": <message>})
instead of status "success"; when `result_obj` is present but contains an error
state propagate that error (include its message/details) and only call
`_extract_text_from_task(result_obj)` and return status "success" when the peer
state and result payload shape are valid.
Comment on lines
+127
to
+134
| async def _sse_stream(message_text: str, context_id: str, task_id: str, rpc_id: Any) -> AsyncIterator[dict]: | ||
| agen = runner.run(message_text, context_id) | ||
| events = [] | ||
| async for ev in agen: | ||
| events.append(ev) | ||
| for upd in stream_events_to_a2a(events, task_id=task_id, context_id=context_id): | ||
| frame = _rpc_result(rpc_id, upd.model_dump(mode="json", exclude_none=True, by_alias=True)) | ||
| yield {"data": json.dumps(frame)} |
Contributor
There was a problem hiding this comment.
Handle stream-time exceptions inside the SSE generator.
If runner.run(...) or stream_events_to_a2a(...) raises, the stream can terminate without a JSON-RPC error frame, leaving clients with an ambiguous broken stream.
Suggested fix
async def _sse_stream(message_text: str, context_id: str, task_id: str, rpc_id: Any) -> AsyncIterator[dict]:
- agen = runner.run(message_text, context_id)
- events = []
- async for ev in agen:
- events.append(ev)
- for upd in stream_events_to_a2a(events, task_id=task_id, context_id=context_id):
- frame = _rpc_result(rpc_id, upd.model_dump(mode="json", exclude_none=True, by_alias=True))
- yield {"data": json.dumps(frame)}
+ try:
+ agen = runner.run(message_text, context_id)
+ events = []
+ async for ev in agen:
+ events.append(ev)
+ for upd in stream_events_to_a2a(events, task_id=task_id, context_id=context_id):
+ frame = _rpc_result(rpc_id, upd.model_dump(mode="json", exclude_none=True, by_alias=True))
+ yield {"data": json.dumps(frame)}
+ except Exception:
+ yield {"data": json.dumps(_rpc_error(rpc_id, _INTERNAL_ERROR, "Internal error"))}Also applies to: 170-178
🤖 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 `@src/cuga/backend/server/a2a/router.py` around lines 127 - 134, The SSE
generator _sse_stream must catch exceptions from runner.run(...) and
stream_events_to_a2a(...) and emit a JSON-RPC error frame instead of just
closing the stream; wrap the run/iteration logic in try/except, build an error
response via _rpc_result(rpc_id, {"error": {...}}) including the exception
message/type (or a sanitized message) and yield it as a JSON string, then
return/stop the generator; apply the same pattern to the other SSE generator
block (the one also using runner.run and stream_events_to_a2a) so any runtime
error produces a single JSON-RPC error frame to clients rather than a silent
broken stream.
Comment on lines
+164
to
+165
| except Exception as exc: # graph blew up | ||
| return JSONResponse(_rpc_error(rpc_id, _INTERNAL_ERROR, "Internal error", str(exc))) |
Contributor
There was a problem hiding this comment.
Do not expose raw exception details in JSON-RPC errors.
Returning str(exc) to remote callers leaks internal implementation details. Keep the wire error generic and log details server-side only.
Suggested fix
- except Exception as exc: # graph blew up
- return JSONResponse(_rpc_error(rpc_id, _INTERNAL_ERROR, "Internal error", str(exc)))
+ except Exception: # graph blew up
+ return JSONResponse(_rpc_error(rpc_id, _INTERNAL_ERROR, "Internal error"))🧰 Tools
🪛 Ruff (0.15.15)
[warning] 164-164: Do not catch blind exception: Exception
(BLE001)
🤖 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 `@src/cuga/backend/server/a2a/router.py` around lines 164 - 165, The handler
currently returns raw exception text via JSONResponse(_rpc_error(rpc_id,
_INTERNAL_ERROR, "Internal error", str(exc))), which leaks internal details;
instead log the full exception server-side (e.g., logging.exception or the
existing logger) and change the JSON-RPC response to return a generic message
only (omit str(exc)), keeping the same rpc_id and _INTERNAL_ERROR constants and
using _rpc_error to build the response; ensure the logged entry includes the
exception and stack for debugging while the wire response contains no internal
exception data.
Comment on lines
+98
to
+107
| if _is_final(ev): | ||
| saw_terminal = True | ||
| yield TaskStatusUpdateEvent( | ||
| task_id=task_id, | ||
| context_id=context_id, | ||
| final=True, | ||
| status=TaskStatus( | ||
| state=TaskState.completed, | ||
| message=_message(_event_text(ev) or "", f"{task_id}-final", context_id), | ||
| ), |
Contributor
There was a problem hiding this comment.
Terminal error events are being marked as completed.
When a final event represents an error (e.g., name "error"), this path still emits TaskState.completed, so failures are reported as success.
Suggested fix
+def _is_error(event: Any) -> bool:
+ name = str(getattr(event, "name", "") or "").lower()
+ return name in {"error", "failed", "failure", "exception"}
+
...
if _is_final(ev):
saw_terminal = True
+ terminal_state = TaskState.failed if _is_error(ev) else TaskState.completed
yield TaskStatusUpdateEvent(
task_id=task_id,
context_id=context_id,
final=True,
status=TaskStatus(
- state=TaskState.completed,
+ state=terminal_state,
message=_message(_event_text(ev) or "", f"{task_id}-final", context_id),
),
)🤖 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 `@src/cuga/backend/server/a2a/task_adapter.py` around lines 98 - 107, The
terminal-event branch in _is_final(...) currently always emits
TaskState.completed even for error terminal events; update the branch that
yields TaskStatusUpdateEvent so it inspects the event (e.g., ev.name or ev.type
via _event_text/ev.name) and chooses TaskState.failed (or the project’s error
state enum) when the final event represents an error, and only uses
TaskState.completed for successful terminals; preserve the existing message
construction via _message(_event_text(ev) ..., f"{task_id}-final", context_id)
and set the TaskStatus.state accordingly in the TaskStatus passed to
TaskStatusUpdateEvent.
Comment on lines
+1632
to
+1719
| if getattr(settings, "a2a", None) and getattr(settings.a2a, "enabled", False): | ||
| # The A2A package is only imported when explicitly enabled in settings, | ||
| # so disabled deployments pay no import-time cost for it. | ||
| from cuga.backend.server.a2a import build_router as _build_a2a_router # noqa: E402 | ||
|
|
||
| class _A2AStreamEvent: | ||
| """Duck-typed event the A2A task adapter consumes.""" | ||
|
|
||
| __slots__ = ("name", "data", "final") | ||
|
|
||
| def __init__(self, name, data=None, final=False): | ||
| self.name = name | ||
| self.data = data | ||
| self.final = final | ||
|
|
||
| class _SupervisorA2ARunner: | ||
| """Run inbound A2A messages through a lazily-created CugaSupervisor. | ||
|
|
||
| We instantiate the supervisor on first use rather than during | ||
| lifespan startup so deployments that enable A2A but never receive | ||
| a request pay no init cost. The supervisor is cached on | ||
| ``app_state.a2a_supervisor`` afterward. | ||
| """ | ||
|
|
||
| def __init__(self, app_state_ref, supervisor_config_path: str): | ||
| self._app_state = app_state_ref | ||
| self._yaml_path = supervisor_config_path | ||
| self._lock = asyncio.Lock() | ||
|
|
||
| async def _ensure_supervisor(self): | ||
| existing = getattr(self._app_state, "a2a_supervisor", None) | ||
| if existing is not None: | ||
| return existing | ||
| async with self._lock: | ||
| existing = getattr(self._app_state, "a2a_supervisor", None) | ||
| if existing is not None: | ||
| return existing | ||
| from cuga.sdk import CugaSupervisor | ||
|
|
||
| supervisor = await CugaSupervisor.from_yaml(self._yaml_path) | ||
| self._app_state.a2a_supervisor = supervisor | ||
| return supervisor | ||
|
|
||
| async def run(self, message, context_id=None): | ||
| try: | ||
| supervisor = await self._ensure_supervisor() | ||
| result = await supervisor.invoke(message, thread_id=context_id) | ||
| answer = getattr(result, "answer", None) or str(result) | ||
| error = getattr(result, "error", None) | ||
| if error: | ||
| yield _A2AStreamEvent("error", {"text": f"Supervisor error: {error}"}, final=True) | ||
| return | ||
| yield _A2AStreamEvent("final_answer", {"text": answer}, final=True) | ||
| except Exception as exc: # surface failures across the wire | ||
| logger.exception("A2A inbound delegation failed") | ||
| yield _A2AStreamEvent("error", {"text": f"A2A handler error: {exc}"}, final=True) | ||
|
|
||
| class _PlaceholderA2ARunner: | ||
| """Used when ``a2a.supervisor_config_path`` is unset. | ||
|
|
||
| Returns a clear "endpoint reached but unconfigured" terminal event | ||
| so callers get a well-formed Task envelope instead of a 5xx. | ||
| """ | ||
|
|
||
| async def run(self, message, context_id=None): | ||
| yield _A2AStreamEvent( | ||
| "final_answer", | ||
| {"text": "A2A inbound endpoint reached, but settings.a2a.supervisor_config_path is not set."}, | ||
| final=True, | ||
| ) | ||
|
|
||
| _a2a_settings_dict = { | ||
| "agent_name": getattr(settings.a2a, "agent_name", "cuga"), | ||
| "agent_description": getattr(settings.a2a, "agent_description", "CUGA agent exposed over A2A."), | ||
| "agent_version": getattr(settings.a2a, "agent_version", "0.0.0"), | ||
| "agent_url": getattr(settings.a2a, "agent_url", "http://localhost:8000"), | ||
| "auth_required": getattr(settings.a2a, "auth_required", False), | ||
| "skill_ids": list(getattr(settings.a2a, "skill_ids", []) or []), | ||
| } | ||
| _a2a_supervisor_cfg = getattr(settings.a2a, "supervisor_config_path", "") or "" | ||
| if _a2a_supervisor_cfg: | ||
| _a2a_runner: Any = _SupervisorA2ARunner(app_state, _a2a_supervisor_cfg) | ||
| logger.info(f"A2A inbound: routing requests through supervisor at {_a2a_supervisor_cfg}") | ||
| else: | ||
| _a2a_runner = _PlaceholderA2ARunner() | ||
| logger.warning("A2A inbound enabled but settings.a2a.supervisor_config_path is empty; using placeholder runner.") | ||
| app.include_router(_build_a2a_router(runner=_a2a_runner, settings=_a2a_settings_dict)) | ||
|
|
Contributor
There was a problem hiding this comment.
Ruff format check is failing for this file.
CI reports this file would be reformatted; run formatter before merge.
🧰 Tools
🪛 Ruff (0.15.15)
[warning] 1640-1640: _A2AStreamEvent.__slots__ is not sorted
Apply a natural sort to _A2AStreamEvent.__slots__
(RUF023)
[warning] 1685-1685: Do not catch blind exception: Exception
(BLE001)
🤖 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 `@src/cuga/backend/server/main.py` around lines 1632 - 1719, Ruff reported that
src/cuga/backend/server/main.py needs reformatting; run the project's formatter
(ruff/black as configured) to reformat this file (or the repo) and fix style
issues around the A2A block (e.g., the classes _A2AStreamEvent,
_SupervisorA2ARunner, _PlaceholderA2ARunner and the
_a2a_settings_dict/_a2a_supervisor_cfg logic), then re-run the linter and commit
the formatted changes so CI passes.
Source: Pipeline failures
Comment on lines
+1685
to
+1687
| except Exception as exc: # surface failures across the wire | ||
| logger.exception("A2A inbound delegation failed") | ||
| yield _A2AStreamEvent("error", {"text": f"A2A handler error: {exc}"}, final=True) |
Contributor
There was a problem hiding this comment.
Do not send raw exception text over A2A error events.
f"A2A handler error: {exc}" can disclose internal paths/config/runtime details to external callers. Send a generic message and keep details in server logs.
Suggested fix
except Exception as exc: # surface failures across the wire
logger.exception("A2A inbound delegation failed")
- yield _A2AStreamEvent("error", {"text": f"A2A handler error: {exc}"}, final=True)
+ yield _A2AStreamEvent("error", {"text": "A2A handler error"}, final=True)🧰 Tools
🪛 Ruff (0.15.15)
[warning] 1685-1685: Do not catch blind exception: Exception
(BLE001)
🤖 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 `@src/cuga/backend/server/main.py` around lines 1685 - 1687, The A2A error
event is leaking raw exception text via f"A2A handler error: {exc}"; in the
except block that catches Exception (where logger.exception(...) and yield
_A2AStreamEvent(...) are called) change the yielded event to include a generic
error message (e.g. "A2A handler error") without including exc or any internal
details, keep the logger.exception call as-is so full exception details remain
in server logs, and ensure the _A2AStreamEvent still sets final=True.
Comment on lines
+87
to
+90
| class _ProtoCard: | ||
| url = "" # absent on protobuf | ||
| supported_interfaces = [_Iface()] | ||
|
|
Contributor
There was a problem hiding this comment.
Use immutable class attributes in test doubles (RUF012).
Line 89 and Line 109 define mutable list class attributes for supported_interfaces. Switching to tuples avoids shared mutable state and clears the Ruff warning.
Suggested fix
class _ProtoCard:
url = "" # absent on protobuf
- supported_interfaces = [_Iface()]
+ supported_interfaces = (_Iface(),)
@@
class _Card:
url = ""
- supported_interfaces = [_Grpc(), _Json()]
+ supported_interfaces = (_Grpc(), _Json())Also applies to: 107-110
🧰 Tools
🪛 Ruff (0.15.15)
[warning] 89-89: Mutable default value for class attribute
(RUF012)
🤖 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/integration/a2a/test_outbound_protocol.py` around lines 87 - 90, Change
mutable list class attributes used in the test doubles to immutable tuples:
replace occurrences of supported_interfaces = [_Iface()] in the _ProtoCard test
double (and the other test double at lines ~107-110) with supported_interfaces =
(_Iface(),) so the class-level attribute is immutable and avoids shared mutable
state and the RUF012 warning; update any other similar class-level list literals
in those test classes to tuple equivalents.
Source: Linters/SAST tools
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
2 participants
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.
Summary
src/cuga/backend/server/a2a/) that exposes CUGA over A2A v0.3 (JSON-RPC) —/.well-known/agent.json,/.well-known/agent-card.json, and a/a2aendpoint supportingmessage/sendand SSEmessage/stream. Mounted only whensettings.a2a.enabled = true, so disabled deployments are byte-identical to before.CugaSupervisorconfigured viasettings.a2a.supervisor_config_path. First request constructs the supervisor, caches it onapp_state.a2a_supervisor, and emits the answer as a singlefinal_answerevent.delegate_task_via_a2a_sdkto speak v0.3 JSON-RPC directly viahttpx. The previous implementation imported names removed ina2a-sdk1.x (A2AClient,MessageSendParams,TextPart,JSONRPCErrorResponse,a2a.utils.message), which madeHAS_A2A_SDK = Falseand silently disabled HTTP-A2A delegation in production..url) and the v1.0 protobuf model thatA2ACardResolver.get_agent_card()actually returns (URL lives atsupported_interfaces[0].urlwith aprotocol_binding). Without this, every cross-process delegation failed with "Agent card carries no URL".docs/examples/a2a_two_cuga/: provider YAML (digital_sales toolset), consumer YAML (no local tools, one external A2A agent), launch scripts, and a preflight that loads.envand works around Saved-config startup wipes MODEL_NAME env var when llm.model is empty (force_env mode) #322.test_outbound_protocol.py(8 tests) covers_extract_text_from_task, the proto-shape AgentCard URL extraction, JSONRPC-vs-GRPC interface preference, end-to-end roundtrip viahttpx.ASGITransport, and JSON-RPC error envelope handling. Existingtest_cuga_to_cuga.py"delegate" test now actually exercises the rewritten outbound client end-to-end.End-to-end verification
Smoke-tested live in this branch:
Result: consumer chat UI → consumer supervisor → A2A JSON-RPC → provider on :8002 → digital_sales tools → 50 real territory accounts → final answer back in the chat UI. Provider log shows
POST /a2a → 200, consumer log showsDelegating to provider: List my territory accounts....Related
MODEL_NAMEenv var whenllm.modelis empty (force_env mode). The demo's_preflight.pyis a workaround for this; once Saved-config startup wipes MODEL_NAME env var when llm.model is empty (force_env mode) #322 lands the preflight can be deleted.Out of scope
xfailtest stays xfail).message/stream— currently folds the whole supervisor run into one terminal frame.cuga_mode,memory_provider, github MCP, drawio MCP) are deliberately not in this PR.Test plan
uv run pytest tests/unit/a2a/ tests/integration/a2a/→ 40 passed, 1 xfaileduv run ruff check src/cuga/backend/server/a2a/ src/cuga/backend/server/main.py src/cuga/backend/cuga_graph/nodes/cuga_supervisor/a2a_protocol.py tests/integration/a2a/→ cleanete-litellmgatewaysettings.a2a.enabled = false)_agent_card_rpc_urlagainst any internal A2A peers that aren't CUGA-frontedSummary by CodeRabbit
New Features
Documentation
Tests