Step 4 · Call the agent over A2A, safely (12 min)¶

Goal: Make the Orchestrator call the SRE agent over A2A, and let it do nothing else.

Builds on: A2A basics: 4. Task, 6. Metadata and extensions, 7. Client. Antigravity SDK basics: 3. Policies.

The idea, part 1: an A2A client¶

The Orchestrator (agent/src/agent/config.py) is the agent that users talk to. It has one tool for each skill on the SRE agent's card:

Orchestrator tool	SRE agent skill
`list_incidents`	`list_incidents` (fast)
`diagnose_sre`	`diagnose_incident`
`write_post_mortem`	`write_post_mortem` (step 3)

All three tools use _call_sre_skill, which calls call_agent(...) from sre_common. call_agent reads the agent card, sends one A2A message, forwards the progress updates, and returns the result.

Why the TODO has A2UI lines. The SRE agent sends UI (A2UI surfaces, step 5) only to a caller that asks for it, and a chat asks in its A2A request. These three lines of the TODO are the client side of A2UI:

metadata["a2uiClientCapabilities"] = A2UI_CLIENT_CAPABILITIES: the components that the chat can draw (its catalog).
extensions=[A2UI_EXTENSION_URI]: turns on the A2UI extension for this request.
sink.a2ui = result.a2ui: keeps the surface that comes back, so that the chat can show it.

They are here because they are part of the A2A request. Without them, the chat shows only text. Step 5 explains A2UI, and changes the other side: the surfaces that the SRE agent builds.

On your laptop, there is no SRE agent service. The Orchestrator starts the SRE agent on a free local port (agent/src/agent/local_sre.py). The calls are real A2A over HTTP, as in production.

The idea, part 2: deny by default¶

The Orchestrator uses the Antigravity SDK. Its agents have powerful built-in tools: they can read files, run commands and get URLs. An on-call chat agent does not need these tools.

The Antigravity runtime applies policies before each tool call. The model cannot change a policy decision:

from google.antigravity.hooks.policy import allow, ask_user, deny

[deny("*"), allow("diagnose_sre")]  # no tools, except diagnose_sre

A rule for a specified tool has priority over the wildcard rule ("*").
deny has priority over ask_user, and ask_user has priority over allow.

At this time, the policy is [deny("*")]. You saw the result in step 0:

The `diagnose_sre` tool call was blocked by the safety policy (decision: deny).

Your task¶

Find the TODOs:
```
git grep -n "TODO(step-4)"
```
In _call_sre_skill, replace the raise with the A2A call. The TODO lists the three parts: the A2UI capabilities, call_agent(...), and the result.
In build_safety_policies(), keep deny("*"), and add one allow(...) for each of the three SRE tools.

Run it¶

Do these commands:

uv run workshop/check.py 4
uv run simulate_incident.py          # the full path, through the Orchestrator: no --engine-only

Find these lines in the log. The Orchestrator starts the SRE agent and calls it over A2A:

Started the SRE agent locally for A2A calls: http://127.0.0.1:...
Calling the SRE agent's diagnose_incident skill over A2A: http://127.0.0.1:...

Start the chat with uv run workshop/chat.py, and open http://localhost:8080/chat.
Ask "What are the latest failures?". Then ask for a post-mortem: your skill from step 3.

The check also runs a contract test against the real Antigravity SDK. The test proves that the SDK receives your policy as {"*": DENY, "list_incidents": ALLOW, "diagnose_sre": ALLOW, "write_post_mortem": ALLOW}.

Discuss¶

Why deny("*") + allow(...), and not only fewer tools? The built-in tools of the SDK are available also when you do not register them. The runtime enforces a policy. An instruction in the system prompt is only a request.
Why one Orchestrator tool for each skill, and not one generic call_sre_agent(skill=...)? Each tool gets its own policy decision. For example, you can ask_user before a post-mortem, and always allow the incident list.
The agent card already lists the skills. Why does the Orchestrator still define its own tools? A skill on a card has no parameter schema: the tools give the model typed arguments, such as trace_id. And the Orchestrator decides what it may do, not the remote agent: a new skill on the card must not give the Orchestrator a new power without a review. You saw this: step 3 put write_post_mortem on the card, and the chat can use it only after this step.

Stretch: human in the loop¶

Add a mock tool restart_service(service_name: str) -> str.
Control the tool with ask_user("restart_service", handler=...).
Write a handler that logs the request and returns False.

This task is Exercise 7 in EXERCISES.md.

Stuck? Run uv run workshop/step.py hint 4, or ask the workshop coach in Antigravity. To apply the solution: uv run workshop/step.py solve 4.

Hint and solution¶

Try the step first. Open the hint when you are stuck, and the solution only after the hint.

Hint: where to change the code

File: agent/src/agent/config.py, in _call_sre_skill(). The TODO says:

# TODO(step-4): Send the request to the SRE agent over A2A, and keep its answer:
#   1. For a chat (`sink is not None`), add A2UI_CLIENT_CAPABILITIES to the metadata, under
#      the key "a2uiClientCapabilities". Then the SRE agent also sends A2UI surfaces.
#   2. result = await call_agent(base_url, prompt, metadata, context_id=..., on_progress=...,
#      extensions=...). Use sink.context_id (or "" without a sink), _emit_progress, and
#      [A2UI_EXTENSION_URI] for a chat (or None).
#   3. report = result.text. For a chat, also keep the surface: sink.a2ui = result.a2ui.

File: agent/src/agent/config.py, in build_safety_policies(). The TODO says:

# TODO(step-4): Everything is denied, so the Orchestrator cannot even delegate. Keep the
#   deny-by-default rule and add one `allow(...)` per tool it needs: one per SRE agent skill.

Solution

--- a/agent/src/agent/config.py
+++ b/agent/src/agent/config.py
@@ -451,15 +451,20 @@
     }
     if trace_id:
         metadata["trace_id"] = trace_id
+    if sink is not None:  # a chat: its UI renders A2UI
+        metadata["a2uiClientCapabilities"] = A2UI_CLIENT_CAPABILITIES
     try:
-        # TODO(step-4): Send the request to the SRE agent over A2A, and keep its answer:
-        #   1. For a chat (`sink is not None`), add A2UI_CLIENT_CAPABILITIES to the metadata, under
-        #      the key "a2uiClientCapabilities". Then the SRE agent also sends A2UI surfaces.
-        #   2. result = await call_agent(base_url, prompt, metadata, context_id=..., on_progress=...,
-        #      extensions=...). Use sink.context_id (or "" without a sink), _emit_progress, and
-        #      [A2UI_EXTENSION_URI] for a chat (or None).
-        #   3. report = result.text. For a chat, also keep the surface: sink.a2ui = result.a2ui.
-        raise NotImplementedError("TODO(step-4): call the SRE agent over A2A")
+        result = await call_agent(
+            base_url,
+            prompt,
+            metadata,
+            context_id=sink.context_id if sink else "",
+            on_progress=_emit_progress,
+            extensions=[A2UI_EXTENSION_URI] if sink is not None else None,
+        )
+        report = result.text
+        if sink is not None:
+            sink.a2ui = result.a2ui
     except Exception as e:
         logger.error(f"Failed to communicate with SRE sub-agent: {e}")
         report = f"Error: Failed to contact SRE Sub-Agent: {e!s}"
@@ -558,9 +563,7 @@

 def build_safety_policies() -> list[Any]:
     """Returns the Orchestrator's tool-call policies: deny everything, allow delegation."""
-    # TODO(step-4): Everything is denied, so the Orchestrator cannot even delegate. Keep the
-    #   deny-by-default rule and add one `allow(...)` per tool it needs: one per SRE agent skill.
-    return [deny("*")]
+    return [deny("*"), allow("list_incidents"), allow("diagnose_sre"), allow("write_post_mortem")]


 def load_agent_config(config_path: str = "agent/agent_config.json") -> LocalAgentConfig:

To apply it to your code: uv run workshop/step.py solve 4.