Skip to content

feat(python): add Blueprint Governed Agent — governance principles via native A2A primitives#536

Open
Michelangelo-Z wants to merge 5 commits intoa2aproject:mainfrom
Michelangelo-Z:feat/blueprint-governed-agent
Open

feat(python): add Blueprint Governed Agent — governance principles via native A2A primitives#536
Michelangelo-Z wants to merge 5 commits intoa2aproject:mainfrom
Michelangelo-Z:feat/blueprint-governed-agent

Conversation

@Michelangelo-Z
Copy link
Copy Markdown

Summary

A minimal Python agent demonstrating how to enforce AI Design Blueprint governance principles using native A2A protocol primitives — no custom middleware, no wrapper framework.

What it demonstrates

Three principles mapped directly to A2A mechanics:

Blueprint Principle A2A Primitive
Explicit approval before destructive actions (P8) TASK_STATE_INPUT_REQUIRED — agent pauses, surfaces confirmation request, resumes on user response
Perceptible background work (P5) TaskStatusUpdateEvent streaming — intermediate working events emitted at each execution step
Mid-task steering — cancellation (P7) cancel() handler + approval-gate flow that honours any non-confirm response

These are the exact three principles the OpenClaw inbox incident violated.

Agent flow

User: "delete /tmp/test.txt"
  ↓
Agent: [input-required] "This will permanently delete the file. Reply 'confirm' to proceed."
  ↓
User: "confirm" (with taskId + contextId)
  ↓
Agent: [working] Confirmed. Validating target path...
Agent: [working] Executing file deletion...
Agent: [completed] File deleted successfully.

— or —

User: "abort"
  ↓
Agent: [canceled] Action aborted — file was not modified.

Quick start

cd samples/python/agents/blueprint_governed_agent/
uv run python __main__.py &      # start agent on :9999
uv run python test_client.py     # run all 3 assertions

Expected output:

AI Design Blueprint — A2A governance validation

Connecting to agent at http://127.0.0.1:9999 ...
Agent card found: Blueprint Governed Agent

  [1] Approval gate... PASS  (state='TASK_STATE_INPUT_REQUIRED', ...)
  [2] Mid-task steering (cancel)... PASS  (state='TASK_STATE_CANCELED')
  [3] Streaming + confirm... PASS  (3 working event(s) → completed, 5 total events)

All 3 assertions passed.

Implementation notes

  • protobuf==5.29.5 pinned — protobuf 7.x removed FieldDescriptor.label from its upb backend, breaking a2a-sdk's internal proto introspection
  • Task re-enqueue on resume — on resume, context.current_task status is reset to TASK_STATE_WORKING before re-enqueueing, to clear stale INPUT_REQUIRED state
  • taskId required for resume — clients must include both contextId and taskId in the resume message

Resources

🤖 Generated with Claude Code

Minimal Python agent demonstrating how to enforce AI Design Blueprint
governance principles using native A2A protocol primitives.

Three principles mapped directly to A2A mechanics:
- P8 (explicit approval before destructive actions) → TASK_STATE_INPUT_REQUIRED
- P5 (perceptible background work) → TaskStatusUpdateEvent streaming
- P7 (mid-task steering/cancellation) → cancel() handler

These are the exact three principles the OpenClaw inbox incident violated.
The example includes a test client that validates all three end-to-end.

See https://aidesignblueprint.com for the full doctrine.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Copy link
Copy Markdown
Contributor

@gemini-code-assist gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces the 'Blueprint Governed Agent' sample, which demonstrates AI Design Blueprint governance principles using the A2A protocol. It includes a server implementation, an executor with approval gates and progress streaming, and a test client. Feedback includes improving the robustness of the confirmation check to prevent accidental execution, correcting directory paths in the documentation to match the project structure, and increasing the logging level for better visibility during development.

return # Pause — wait for user response.

# ── Resume: check confirmation (Blueprint P7) ──
if "confirm" not in user_input.lower():
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security-medium medium

The check "confirm" not in user_input.lower() is potentially unsafe for a governance-focused agent. If a user replies with "do not confirm", the agent will incorrectly proceed because the substring "confirm" is present. It is safer to use a strict equality check to ensure explicit approval as per the instructions provided to the user.

Suggested change
if "confirm" not in user_input.lower():
if user_input.lower() != "confirm":

## Quick start

```bash
cd a2a/
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

low

The directory name a2a/ in the quick start instructions does not match the actual directory path samples/python/agents/blueprint_governed_agent/. This should be updated to avoid confusion for users trying to run the sample.

Suggested change
cd a2a/
cd samples/python/agents/blueprint_governed_agent/

## Files

```
a2a/
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

low

The directory name in the file tree visualization should match the actual directory name in the repository.

Suggested change
a2a/
blueprint_governed_agent/

app = Starlette(routes=routes)

if __name__ == "__main__":
uvicorn.run(app, host=HOST, port=PORT, log_level="warning")
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

low

Using log_level="warning" might hide useful information for users running this sample for the first time. Setting it to "info" allows them to see incoming requests and server activity, which is helpful for debugging and understanding the A2A flow.

Suggested change
uvicorn.run(app, host=HOST, port=PORT, log_level="warning")
uvicorn.run(app, host=HOST, port=PORT, log_level="info")

Michelangelo-Z and others added 4 commits April 25, 2026 21:00
- agent_executor: strict equality check (== "confirm") instead of
  substring match to prevent "do not confirm" from accidentally proceeding
- __main__: log_level "info" so users see request activity on first run
- README: fix quick start path (a2a/ → samples/python/agents/blueprint_governed_agent/)
- README: fix files section directory name to match repo structure

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- add __init__.py to resolve INP001 (implicit namespace package)
- add docstring to execute() to resolve D102
- add # noqa: S101 to test assertions (expected pattern for test clients)
- add # noqa: BLE001 to broad connectivity-check exception (intentional)
- fix import sort order and quote style (ruff format)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Replace British 'honour'/'honours' with American 'honor'/'honors'
in agent_executor.py, test_client.py, and README.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- MD060: add spaces around table separator pipes (|---|---| → | --- | --- |)
- MD040: add language tag to all fenced code blocks (bash/text)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant