Model Context Protocol

MCP is a standard for exposing tools to agent runtimes. AGP is the governance layer that sits underneath those tools. Every call — regardless of which model, framework, or runtime is driving it — gets a capability check, a policy decision, and an immutable audit entry.

AGP is model-agnostic. It does not matter whether the MCP client is Claude, GPT-4, Gemini, a LangChain agent, an AutoGen workflow, or a custom runtime you built yourself. The governance contract is between the tool server and the AGP server — not between AGP and any specific model.

There are two integration points. Choose based on whether you want the agent runtime to participate in the governance workflow, or whether you want governance to fire transparently behind your tool surface.

Approach A — AGP as an MCP server

The agp-mcp package is a ready-made MCP server that exposes the full AGP lifecycle as MCP tools. Any MCP-compatible agent runtime can connect to it — the agent can then register tasks, evaluate policy, and submit action envelopes as native tool calls, with no direct HTTP calls to the AGP server required.

Install

pip install agp-mcp

Start the server

Two transport modes — stdio for local agent runtimes, SSE for remote or networked hosts:

# 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

Connect your agent runtime

Point any MCP-compatible client at the server. Examples for common runtimes:

## Claude Desktop — claude_desktop_config.json
{
  "mcpServers": {
    "agp": {
      "command": "python",
      "args": ["-m", "agp_mcp"],
      "env": { "AGP_BASE_URL": "https://your-agp-server", ... }
    }
  }
}

## LangChain — using langchain-mcp-adapters
from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    "agp": {"url": "http://localhost:8765/sse", "transport": "sse"}
})
tools = await client.get_tools()  # AGP tools available to any LangChain agent

## Any MCP client — stdio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

server_params = StdioServerParameters(
    command="python", args=["-m", "agp_mcp"],
    env={"AGP_BASE_URL": "https://your-agp-server", ...}
)
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()  # all AGP tools available

Available tools

Simple governed action — one tool call

For straightforward low-risk actions, any agent runtime can use run_governed_task to complete the full AGP lifecycle in a single tool call:

High-risk action with human approval

For high-risk actions, the agent runtime uses the individual tools to walk the pipeline step by step, pausing at APPROVAL_PENDING until a human approves. The agent can surface the task ID to the user and wait:

# 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.

Remote / SSE transport

For use with a remote MCP host (not Claude Desktop), start the server in SSE mode:

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

Approach B — AGP behind your MCP server

If you already have an MCP server exposing tools, wrap each handler with the AGP pipeline. The agent runtime calls your tools exactly as before — governance fires server-side, invisibly. The agent sees a normal tool result, or a ToolError if the action was blocked or requires human approval. No changes to the agent's prompts, tool definitions, or client configuration.

Install

pip install agp-sdk mcp

Risk classification

Define how your tools map to AGP risk tiers. This determines whether a tool call auto-approves (allow) or requires human sign-off (require_approval):

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"

Governance wrapper

Wrap your tool handler with a helper that runs the full AGP pipeline before the tool executes. Low and medium risk tools complete in a single round-trip. High-risk tools return an APPROVAL_PENDING response; the real tool only fires once create_approval is called and the wrapper is resumed.

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
    ...

What you get

Next steps

AGP works across the full agent stack. If Claude is orchestrating other agents or calling microservices, governance extends there too: