Model Context Protocol

MCP client  (any model or agent runtime)
          ↓  tool call: "approve_payment"
  AGP governance layer
    ├─ Register task (risk_tier: high)
    ├─ Check capability token
    ├─ Run policy evaluation  →  require_approval
    ├─ Wait for human sign-off
    └─ Submit action envelope  →  201 + ledger entry
          ↓
Tool executes  (or is blocked if any step fails)

pip install agp-mcp

# stdio — for local runtimes (Claude Desktop, local LangChain agent, etc.)
AGP_BASE_URL=https://your-agp-server \
AGP_CLIENT_ID=my-agent \
AGP_CLIENT_SECRET=s3cr3t \
python -m agp_mcp

# SSE — for remote MCP hosts, multi-agent networks, or cloud deployments
AGP_BASE_URL=https://your-agp-server \
AGP_CLIENT_ID=my-agent \
AGP_CLIENT_SECRET=s3cr3t \
python -m agp_mcp --sse

# Agent runtime calls these tools in sequence

# 1. Register intent
create_task(
    principal_id="finance-agent",
    requested_outcome="Approve $125k payment to Acme Corp",
    risk_tier="high",
    sponsoring_entity="finance-team",
    jurisdictions=["EU"],
    regulatory_frameworks=["EU_AI_ACT_2024"],
)
# → { "task_id": "tsk_01HX...", "status": "CREATED" }

# 2. Bind accountability
bind_liability(task_id="tsk_01HX...", sponsoring_entity="finance-team",
               accountable_owner="cfo@acme.com", jurisdiction="EU")

# 3. Issue capability token
issue_capability(subject_agent="finance-agent", issuer="finance-team",
                 principal_id="finance-agent",
                 permitted_actions=["approve_payment"])
# → { "capability_id": "cap_01HA..." }

# 4. Record reasoning
record_decision(task_id="tsk_01HX...", agent_id="finance-agent",
                selected_action="approve_payment",
                rationale="Vendor verified, amount within delegated limit",
                uncertainty_score=0.12)

# 5. Policy evaluation → high-risk triggers require_approval
evaluate_policy(task_id="tsk_01HX...", verdict="require_approval",
                rationale="Amount > $100k threshold",
                decision_ref="dec_01HB...")
# → task status: APPROVAL_PENDING

# Agent reports back to user: "Action registered, awaiting CFO approval.
# Task ID: tsk_01HX... Resume once approved."

# 6. After human approves out-of-band (via your approval UI or AGP API):
create_approval(task_id="tsk_01HX...", approver_id="cfo@acme.com",
                decision="approved", rationale="Confirmed with procurement",
                policy_evaluation_ref="pol_01HC...")

# 7. Execute — fail-closed gate fires here
submit_action(task_id="tsk_01HX...", agent_id="finance-agent",
              tool_id="approve_payment",
              operation={"vendor": "Acme Corp", "amount_usd": 125000},
              capability_token_ref="cap_01HA...",
              decision_ref="dec_01HB...",
              policy_decision_ref="pol_01HC...")
# → 201 Created. Signed receipt + immutable ledger entry.

AGP_BASE_URL=https://your-agp-server \
AGP_CLIENT_ID=my-agent \
AGP_CLIENT_SECRET=s3cr3t \
python -m agp_mcp --sse

pip install agp-sdk mcp

from agp import AGPClient

RISK_MAP = {
    # Read-only, no side effects
    "search_docs":      "low",
    "read_database":    "low",
    # Writes to internal systems
    "send_email":       "medium",
    "update_record":    "medium",
    # Consequential external actions
    "approve_payment":  "high",
    "delete_records":   "high",
    "deploy_code":      "critical",
}

def policy_verdict(risk_tier: str) -> str:
    return "allow" if risk_tier in ("low", "medium") else "require_approval"

import functools
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from agp import AGPClient
from agp.exceptions import AGPPolicyDeniedError, AGPApprovalRequiredError

agp = AGPClient("https://your-agp-server",
                client_id="my-mcp-server", client_secret="s3cr3t")
server = FastMCP("my-tools")


def governed(tool_name: str, agent_id: str = "mcp-server",
             entity: str = "my-org", owner: str = "owner@example.com"):
    """Decorator: wrap any tool handler with AGP governance."""
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            risk = RISK_MAP.get(tool_name, "medium")
            verdict = policy_verdict(risk)

            try:
                with agp.task_session(
                    principal_id=agent_id,
                    requested_outcome=f"MCP tool call: {tool_name}",
                    risk_tier=risk,
                ) as session:
                    session.bind(sponsoring_entity=entity,
                                 accountable_owner=owner, jurisdiction="EU")
                    session.issue_capability(subject_agent=agent_id, issuer=entity,
                                             principal_id=agent_id,
                                             permitted_actions=[tool_name])
                    session.decide(agent_id=agent_id, selected_action=tool_name,
                                   rationale="MCP tool call", uncertainty_score=0.1)
                    session.evaluate(verdict=verdict)

                    if verdict == "require_approval":
                        # Return task_id — caller must approve before execution
                        raise ToolError(
                            f"Action requires human approval. "
                            f"Task ID: {session.task_id}. "
                            f"Approve via POST /agp/decision/approvals."
                        )

                    session.execute(agent_id=agent_id, tool_id=tool_name,
                                    operation=kwargs)
                    return await fn(*args, **kwargs)

            except AGPPolicyDeniedError as e:
                raise ToolError(f"Blocked by governance policy: {e}")
        return wrapper
    return decorator


# Apply to your tools — Claude sees no difference
@server.tool()
@governed("send_email")
async def send_email(to: str, subject: str, body: str) -> str:
    """Send an email."""
    # Your real implementation here
    ...

@server.tool()
@governed("approve_payment")
async def approve_payment(vendor: str, amount_usd: float) -> str:
    """Approve a vendor payment."""
    # Only reached after AGP clears the action envelope
    ...