MCP Integration

Integrate ACGP with the Model Context Protocol (MCP) to govern tool calls and data access before execution.

Why Govern MCP Calls

MCP extends governed agent behavior through tools and external systems. Governance is required to: - Validate requested tool/action against policy - Enforce Governance Tier and scope constraints - Prevent unsafe parameter patterns - Audit every high-impact tool call

Installation

pip install "acgp-sdk[mcp,postgres]"

Basic Adapter Setup

from acgp import GovernanceSteward, PostgresStateStorage
from acgp.integrations.mcp import MCPGovernanceAdapter

steward = GovernanceSteward.production(
    blueprint_file="blueprint.yaml",
    state_storage=PostgresStateStorage(connection_string="postgresql://runtime/acgp"),
)

governed_mcp = MCPGovernanceAdapter(
    mcp_server=my_mcp_server,
    steward=steward,
)

Governed Tool Invocation

result = governed_mcp.call_tool(
    tool="file_system_write",
    parameters={"path": "/data/file.txt", "content": "..."},
)

Before execution, the adapter should: 1. Build a cognitive trace from tool name + parameters + context. 2. Evaluate with GovernanceSteward. 3. Enforce intervention decision and optional flag workflow. 4. Persist audit metadata (trace_id, tool, decision, reason).

Blueprint Pattern for MCP Tools

tripwires:
  - id: block_sensitive_write_paths
    when:
      hook: tool_call
      tool: file_system_write
    condition: "args.path contains '/etc/'"
    eval_tier: 0
    on_fail:
      decision: block
      reason: "Sensitive system path write attempted"

checks:
  - id: tool_argument_sanity
    when:
      hook: tool_call
      tool: file_system_write
    metric:
      name: arg_safety
      weight: 1.0
      check:
        type: rule-based

Intervention Handling Pattern

eval_result = steward.evaluate(trace)

if eval_result.intervention in {"ok", "nudge"}:
    output = mcp_server.call_tool(tool, parameters)
elif eval_result.intervention == "escalate":
    if request_human_approval(trace, eval_result.message):
        output = mcp_server.call_tool(tool, parameters)
    else:
        raise PermissionError("Tool call denied pending approval")
elif eval_result.intervention == "block":
    raise PermissionError(eval_result.message)
else:  # halt
    raise RuntimeError("Agent halted by governance policy")

if eval_result.flags and eval_result.flags.flagged:
    queue_tool_audit(trace_id=eval_result.trace_id)

Security Notes

  • Never allow unchecked passthrough to privileged tools.
  • Maintain per-tool allowlists by agent role.
  • Validate parameter schema before policy evaluation and execution.
  • Use immutable audit storage for high-impact tool actions.

Performance Notes

  • Keep frequent low-risk tool checks in Tier 0/1.
  • Use cached policy context for repetitive tool categories.
  • Monitor timeout/fallback rates for tool-heavy workloads.