ACGP-1008: Interoperability Specification¶

Status: Draft
Last Updated: 2026-01-08
Spec ID: ACGP-1008
Normative Keywords: MUST, SHOULD, MAY (per RFC 2119)

Abstract¶

This document specifies the interoperability mechanisms between the Agentic Cognitive Governance Protocol (ACGP) and external agent protocols, specifically the Model Context Protocol (MCP) for tool usage and the Agent-to-Agent (A2A) Protocol for inter-agent communication. It defines adapter patterns, message translation requirements, governance injection points, and security considerations for seamless integration while maintaining ACGP's governance guarantees. This specification enables ACGP to govern agents regardless of their underlying tool or communication protocols.

Table of Contents¶

Introduction
MCP (Model Context Protocol) Integration
A2A (Agent-to-Agent Protocol) Integration
Common Adapter Patterns
Message Translation and Mapping
Governance Injection Points
Security Considerations
Performance Requirements
Reference Implementations
Conformance Requirements
References

1. Introduction¶

Modern AI agents operate in heterogeneous environments using various protocols for tool access and inter-agent communication. ACGP must integrate seamlessly with these protocols while maintaining its governance guarantees. This specification defines how ACGP acts as a governance layer above existing protocols without requiring modifications to those protocols.

1.1 Integration Principles¶

Non-invasive: External protocols continue functioning without modification
Transparent: Governance is invisible to compliant operations
Performant: Minimal latency overhead (<10ms P95)
Secure: No weakening of existing security boundaries
Auditable: Complete trace of all cross-protocol interactions

1.2 Requirements Language¶

The key words "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT", "SHOULD", "SHOULD NOT", "RECOMMENDED", "MAY", and "OPTIONAL" in this document are to be interpreted as described in RFC 2119.

2. MCP (Model Context Protocol) Integration¶

2.1 MCP Overview¶

The Model Context Protocol enables standardized tool usage across different AI models and platforms. ACGP integrates with MCP by intercepting tool requests and responses, applying governance before allowing tool execution.

2.2 Integration Architecture¶

graph LR
    subgraph "Agent Layer"
        PA[Operating Agent]
        ACGP[ACGP Adapter]
    end

    subgraph "MCP Layer"
        MC[MCP Client]
        MS[MCP Server]
    end

    subgraph "Tool Layer"
        T1[Tool 1]
        T2[Tool 2]
        T3[Tool 3]
    end

    PA -->|1. Request| ACGP
    ACGP -->|2. Evaluate + Check Tripwires| ACGP
    ACGP -->|3. Forward if OK| MC
    MC -->|4. Protocol| MS
    MS -->|5. Execute| T1
    T1 -->|6. Result| MS
    MS -->|7. Response| MC
    MC -->|8. Return| ACGP
    ACGP -->|9. Log & Return| PA

    style ACGP fill:#667eea,color:#fff

2.3 MCP Tool Request Interception¶

2.3.1 Pre-execution Governance¶

Before an MCP tool call is executed:

class MCPGovernanceAdapter:
    def intercept_tool_request(self, request: MCPToolRequest) -> MCPResponse:
        # Step 1: Convert MCP request to Cognitive Trace
        trace = self.mcp_to_trace(request)

        # Step 2: Apply ACGP governance
        intervention = self.governance_steward.evaluate(trace)

        # Step 3: Handle intervention (six types)
        if intervention.decision == "ok":
            return self.forward_to_mcp(request)
        elif intervention.decision == "nudge":
            modified_request = self.apply_modifications(request, intervention)
            return self.forward_to_mcp(modified_request)
        elif intervention.decision == "escalate":
            return self.await_human_approval(request)
        elif intervention.decision in ["block", "halt"]:
            return MCPResponse(
                success=False,
                error=f"Blocked by governance: {intervention.reason}"
            )

        # Handle flag (orthogonal to primary intervention)
        if intervention.flagged:
            self.update_trust_debt(intervention.flag_severity)

2.3.2 MCP to Cognitive Trace Mapping¶

{
  "mcp_request": {
    "tool": "database_query",
    "parameters": {
      "query": "SELECT * FROM users",
      "database": "production"
    },
    "context": {
      "session_id": "abc123",
      "user": "analyst_1"
    }
  },
  "cognitive_trace": {
    "trace_id": "uuid-v4",
    "session_id": "abc123",
    "tool_calls": [{
      "name": "database_query",
      "args": {
        "query": "SELECT * FROM users",
        "database": "production"
      }
    }],
    "meta": {
      "protocol": "mcp",
      "mcp_context": {...},
      "acl_tier": "ACL-3"
    }
  }
}

2.4 Tool Categories and Default Governance¶

MCP tools MUST be categorized for appropriate governance:

Category	Risk Level	Default ACL	Example Tools	Governance
Read-Only	Low	ACL-0+	search, read_file, get_weather	Basic logging
Limited Write	Medium	ACL-1+	create_draft, send_notification	Approval for sensitive
External API	Medium	ACL-2+	http_request, api_call	Rate limiting, cost caps
Database	High	ACL-3+	sql_execute, nosql_query	Query validation
System	Critical	ACL-4+	shell_execute, file_delete	Pre-approval required
Financial	Critical	ACL-5	transfer_funds, place_order	Dual approval, limits

2.5 MCP-Specific Tripwires¶

Additional tripwires for MCP tool usage:

mcp_tripwires:
  - id: production_write
    category: critical
    severity: critical
    condition: |
      tool_name IN ['database_write', 'api_post'] AND
      environment == 'production'
    action: 
      acl_0_2: escalate
      acl_3_5: block

  - id: bulk_operation
    category: standard
    severity: standard
    condition: |
      affected_records > 1000 OR
      batch_size > 100
    action: escalate

  - id: credential_access
    category: critical
    severity: severe
    condition: |
      tool_name == 'get_secret' OR
      response.contains(['password', 'api_key', 'token'])
    action: halt

2.6 Post-execution Validation¶

After MCP tool execution:

def validate_tool_response(self, response: MCPToolResponse) -> MCPToolResponse:
    # Check for sensitive data in response
    if self.contains_sensitive_data(response):
        if self.agent.acl_tier <= 2:
            response = self.redact_sensitive_data(response)
        else:
            self.trigger_escalation("Sensitive data in tool response")

    # Verify response matches expected schema
    if not self.validate_schema(response):
        self.flag_anomaly("Unexpected response format")

    # Log the complete interaction
    self.reflection_db.log({
        "type": "mcp_tool_execution",
        "request": request,
        "response": response,
        "governance": intervention,
        "validations": validation_results
    })

    return response

3. A2A (Agent-to-Agent Protocol) Integration¶

3.1 A2A Overview¶

The Agent-to-Agent Protocol enables autonomous agents to discover, negotiate, and collaborate. ACGP governs these interactions by monitoring communications, validating message content, and preventing unauthorized information sharing.

3.2 A2A Governance Architecture¶

sequenceDiagram
    participant A1 as Agent 1
    participant GS1 as Governance Steward 1
    participant A2A as A2A Network
    participant GS2 as Governance Steward 2
    participant A2 as Agent 2
    participant SN as Steward Network

    Note over A1,A2: Agent Discovery Phase
    A1->>GS1: Request to connect
    GS1->>GS1: Verify A2 credentials
    GS1->>SN: Query A2 reputation
    SN-->>GS1: Reputation score

    alt Approved
        GS1->>A2A: Allow connection
        A2A->>GS2: Connection request
        GS2->>GS2: Verify A1 credentials
        GS2->>A2: Connection available
    else Blocked
        GS1->>A1: Connection denied
    end

    Note over A1,A2: Message Exchange Phase
    A1->>GS1: Outbound message
    GS1->>GS1: Content validation + Tripwire check
    GS1->>A2A: Forward message
    A2A->>GS2: Deliver message
    GS2->>GS2: Inbound validation + Tripwire check
    GS2->>A2: Deliver if safe

3.3 A2A Message Governance¶

3.3.1 Agent Identity Verification¶

{
  "agent_card": {
    "id": "agent-uuid",
    "name": "financial-analyst",
    "capabilities": ["data_analysis", "report_generation"],
    "governance": {
      "agp_enabled": true,
      "agp_version": "1.0.2",
      "acl_tier": "ACL-3",
      "blueprints": ["finance/analysis@2.0", "clarity.baseline@1.0"],
      "trust_score": 0.85,
      "trust_debt": 0.2,
      "last_assessment": "2025-01-15T10:00:00Z"
    }
  }
}

3.3.2 Message Content Filtering¶

Outbound and inbound messages MUST be filtered based on ACL tier:

ACL Tier	Outbound Restrictions	Inbound Restrictions
ACL-0	Public data only	No execution requests
ACL-1	Internal data allowed	Simple requests only
ACL-2	Sensitive data with approval	Validated sources
ACL-3	Encrypted sensitive data	Trust score > 0.7
ACL-4	Classified with audit	Known agents only
ACL-5	Dual approval required	Pre-approved list

3.4 Collaborative Task Governance¶

When agents collaborate via A2A:

class A2ACollaborationGovernor:
    def validate_collaboration_request(self, request: A2ATaskRequest):
        # Check if collaboration is allowed
        if not self.is_collaboration_allowed(request):
            return self.block_collaboration(request)

        # Verify task alignment with blueprints
        alignment_score = self.calculate_alignment(
            request.task,
            self.agent.blueprints
        )

        if alignment_score < 0.8:
            return self.escalate_for_review(request)

        # Check tripwires for collaborative tasks
        tripwire_result = self.check_collaboration_tripwires(request)
        if tripwire_result:
            return self.handle_tripwire(tripwire_result)

        # Set collaboration boundaries
        boundaries = self.define_boundaries(request)
        return A2ATaskResponse(
            accepted=True,
            boundaries=boundaries,
            monitoring_level="enhanced"
        )

    def define_boundaries(self, request):
        return {
            "data_sharing": self.get_data_sharing_policy(),
            "resource_limits": self.get_resource_caps(),
            "time_limit": self.calculate_time_limit(),
            "escalation_triggers": self.get_escalation_rules(),
            "termination_conditions": self.get_termination_rules()
        }

3.5 A2A-Specific Security Controls¶

3.5.1 Message Authentication¶

All A2A messages MUST be signed using ES256 and verified:

def sign_a2a_message(self, message: dict) -> dict:
    # Add governance metadata
    message["governance"] = {
        "sender_acl": self.agent.acl_tier,
        "trust_debt": self.agent.trust_debt,
        "blueprint_version": self.blueprint.version,
        "agp_version": "1.0.2",
        "timestamp": datetime.utcnow().isoformat()
    }

    # Create ES256 signature
    signature = self.crypto.sign_es256(
        json.dumps(message, sort_keys=True),
        self.agent.private_key
    )

    message["signature"] = {
        "algorithm": "ES256",
        "value": signature,
        "key_id": self.agent.key_id
    }

    return message

3.5.2 Collusion Detection¶

Monitor for suspicious coordination patterns:

collusion_tripwires:
  - id: rapid_message_exchange
    severity: standard
    threshold: 100 messages in 60 seconds
    action: flag

  - id: data_aggregation
    severity: critical
    condition: multiple agents requesting same sensitive data
    window: 5 minutes
    action: escalate

  - id: coordinated_actions
    severity: critical
    pattern: similar tool calls across agents
    correlation: > 0.9
    action: block

  - id: circular_delegation
    severity: severe
    depth: > 3 agents
    action: halt

4. Common Adapter Patterns¶

4.1 Proxy Pattern¶

Intercept all protocol messages through a governance proxy:

class GovernanceProxy:
    def __init__(self, target_protocol, governance_steward):
        self.protocol = target_protocol
        self.steward = governance_steward

    def execute(self, request):
        # Pre-execution governance
        trace = self.to_cognitive_trace(request)
        intervention = self.steward.evaluate(trace)

        if not self.is_allowed(intervention):
            return self.handle_blocked(request, intervention)

        # Execute with monitoring
        with self.monitor_context():
            response = self.protocol.execute(request)

        # Post-execution validation
        validated_response = self.validate_response(response)

        # Log complete interaction
        self.log_interaction(request, response, intervention)

        return validated_response

4.2 Middleware Pattern¶

Inject governance as middleware in the protocol stack:

class ACGPMiddleware:
    def __init__(self, next_handler):
        self.next = next_handler
        self.steward = GovernanceSteward()

    async def handle(self, context):
        # Before processing
        await self.before_process(context)

        # Process with governance
        if self.should_proceed(context):
            result = await self.next.handle(context)
        else:
            result = self.block_request(context)

        # After processing
        await self.after_process(context, result)

        return result

4.3 Wrapper Pattern¶

Wrap existing protocol clients with governance:

class GovernedMCPClient(MCPClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.governor = MCPGovernanceAdapter()

    def call_tool(self, tool_name, parameters):
        # Governance check (including tripwires)
        if not self.governor.pre_check(tool_name, parameters):
            raise GovernanceException("Tool call blocked by tripwire")

        # Original functionality with monitoring
        with self.governor.monitor():
            result = super().call_tool(tool_name, parameters)

        # Post-processing
        return self.governor.post_process(result)

5. Message Translation and Mapping¶

5.1 Universal Translation Schema¶

All external protocol messages MUST be translatable to Cognitive Traces:

{
  "universal_schema": {
    "protocol": "mcp|a2a|custom",
    "protocol_version": "string",
    "action_type": "tool_call|message|query|command",
    "actor": {
      "id": "agent-uuid",
      "acl_tier": "ACL-X",
      "trust_debt": 0.0,
      "context": {}
    },
    "target": {
      "type": "tool|agent|service",
      "id": "target-identifier",
      "risk_level": "low|medium|high|critical"
    },
    "payload": {
      "original": {},
      "normalized": {}
    },
    "metadata": {
      "timestamp": "iso-8601",
      "session_id": "uuid",
      "correlation_id": "uuid"
    }
  }
}

5.2 Protocol-Specific Mappings¶

5.2.1 MCP Mapping Table¶

MCP Field	Cognitive Trace Field	Transform
`tool`	`tool_calls[0].name`	Direct
`parameters`	`tool_calls[0].args`	JSON copy
`context.session`	`session_id`	Direct
`context.user`	`meta.user_id`	Direct
`timeout`	`meta.timeout_ms`	Seconds to ms

5.2.2 A2A Mapping Table¶

A2A Field	Cognitive Trace Field	Transform
`message.content`	`outputs.text`	Direct
`sender.id`	`meta.sender_agent`	Direct
`task.goal`	`inputs.goal`	Direct
`negotiation.terms`	`meta.collaboration`	Structured
`reputation`	`meta.trust_score`	Normalized

6. Governance Injection Points¶

6.1 Critical Injection Points¶

Governance MUST be enforced at these points:

flowchart LR
    subgraph "Injection Points"
        P1[Protocol Connection]
        P2[Authentication]
        P3[Version Negotiation]
        P4[Pre-execution]
        P5[Execution Monitoring]
        P6[Post-execution]
        P7[Response Filtering]
        P8[Disconnection]
    end

    P1 -->|Verify| P2
    P2 -->|Negotiate| P3
    P3 -->|Authorize| P4
    P4 -->|Evaluate + Tripwires| P5
    P5 -->|Monitor| P6
    P6 -->|Validate| P7
    P7 -->|Clean| P8

    style P4 fill:#ff9999
    style P6 fill:#ff9999

6.2 Injection Point Specifications¶

Point	Requirement	Enforcement	Fallback
Connection	Verify agent identity	REQUIRED	Deny connection
Authentication	Validate credentials	REQUIRED	Terminate
Version Negotiation	Establish protocol version	REQUIRED	Use default
Pre-execution	Governance evaluation + tripwires	REQUIRED	Block action
Monitoring	Runtime observation	RECOMMENDED	Log warning
Post-execution	Response validation	REQUIRED	Quarantine response
Filtering	Sensitive data removal	REQUIRED	Redact all
Disconnection	Clean shutdown	RECOMMENDED	Force terminate

7. Security Considerations¶

7.1 Protocol-Level Security¶

7.1.1 Transport Security¶

All inter-protocol communication MUST use: - TLS 1.3 or higher for network transport - Mutual TLS for service authentication - Perfect forward secrecy - Certificate pinning for known services

7.1.2 Message Security¶

class MessageSecurity:
    def secure_message(self, message, protocol):
        # Encrypt sensitive fields
        if self.has_sensitive_data(message):
            message = self.encrypt_sensitive_fields(message)

        # Add integrity check (SHA-256)
        message["checksum"] = self.calculate_checksum_sha256(message)

        # Sign message (ES256 - standardized)
        message["signature"] = self.sign_message_es256(message)

        # Add security headers
        message["security"] = {
            "algorithm": "ES256",
            "key_id": self.key_id,
            "timestamp": datetime.utcnow().isoformat(),
            "nonce": secrets.token_urlsafe(16)
        }

        return message

7.2 Cross-Protocol Security Risks¶

Risk	Description	Mitigation
Protocol Confusion	Mixing protocols in same session	Strict protocol isolation
Privilege Escalation	Using A2A to bypass MCP limits	Unified permission model
Data Leakage	Information flow between protocols	Data classification enforcement
Replay Attacks	Reusing valid messages	Nonce and timestamp validation
Man-in-the-Middle	Intercepting protocol messages	End-to-end encryption (TLS 1.3)

7.3 Security Monitoring¶

security_monitoring:
  protocol_anomalies:
    - unusual_protocol_mixing
    - rapid_protocol_switching
    - unauthorized_protocol_bridge

  authentication_events:
    - failed_authentication_spike
    - credential_stuffing_attempt
    - token_reuse_detected

  data_flow_analysis:
    - cross_protocol_data_transfer
    - sensitive_data_protocol_mismatch
    - unexpected_data_aggregation

8. Performance Requirements¶

8.1 Latency Overhead¶

Governance MUST NOT add excessive latency:

Operation	Maximum Overhead	P95 Target
Protocol handshake	10ms	25ms total
Version negotiation	5ms	15ms total
Pre-execution check	20ms	50ms total
Message translation	5ms	10ms total
Tripwire evaluation	10ms	20ms total
Post-execution validation	10ms	25ms total
End-to-end governance	50ms	150ms total

8.2 Throughput Requirements¶

Protocol	Messages/sec	Governance Overhead
MCP Tool Calls	1000	<5% degradation
A2A Messages	5000	<10% degradation
Hybrid Operations	500	<15% degradation

8.3 Resource Utilization¶

resource_limits:
  governance_adapter:
    cpu_overhead: <10%
    memory_overhead: <100MB
    network_overhead: <5%

  message_buffer:
    max_size: 10000 messages
    ttl: 60 seconds
    overflow_policy: drop_oldest

  connection_pool:
    max_connections: 1000
    connection_timeout: 30s
    idle_timeout: 300s

9. Reference Implementations¶

9.1 MCP Adapter Example¶

from ACGP import GovernanceSteward, CognitiveTrace
from mcp import MCPClient

class ACGPMCPAdapter:
    def __init__(self, mcp_client: MCPClient, blueprint: str):
        self.mcp = mcp_client
        self.steward = GovernanceSteward(blueprint)
        self.acl_tier = self.steward.get_acl_tier()

    async def call_tool(self, tool_name: str, params: dict):
        # Create trace
        trace = CognitiveTrace(
            tool_calls=[{
                "name": tool_name,
                "args": params
            }],
            meta={
                "protocol": "mcp",
                "acl_tier": self.acl_tier
            }
        )

        # Get governance decision
        intervention = await self.steward.evaluate(trace)

        # Handle intervention (six types)
        if intervention.decision == "ok":
            result = await self.mcp.call_tool(tool_name, params)
        elif intervention.decision == "nudge":
            params = intervention.modified_params
            result = await self.mcp.call_tool(tool_name, params)
        elif intervention.decision == "escalate":
            approval = await self.request_human_approval()
            if approval:
                result = await self.mcp.call_tool(tool_name, params)
            else:
                raise BlockedException("Human denied request")
        else:  # BLOCK or HALT
            raise BlockedException(f"Governance: {intervention.reason}")

        # Handle flag (orthogonal)
        if intervention.flagged:
            await self.update_trust_debt(intervention.flag_severity)

        # Log and return
        await self.log_execution(trace, intervention, result)
        return result

9.2 A2A Adapter Example¶

from ACGP import GovernanceSteward, CognitiveTrace
from a2a import A2AClient

class ACGPA2AAdapter:
    def __init__(self, a2a_client: A2AClient, blueprint: str):
        self.a2a = a2a_client
        self.steward = GovernanceSteward(blueprint)
        self.message_filter = MessageFilter(self.steward.acl_tier)

    async def send_message(self, recipient: str, message: dict):
        # Filter outbound message
        filtered = self.message_filter.filter_outbound(message)

        # Create trace for governance
        trace = CognitiveTrace(
            outputs={"message": filtered},
            meta={
                "protocol": "a2a",
                "recipient": recipient,
                "acl_tier": self.steward.acl_tier
            }
        )

        # Evaluate
        intervention = await self.steward.evaluate(trace)

        if intervention.is_allowed():
            # Sign with ES256 and send
            signed = self.sign_message_es256(filtered)
            response = await self.a2a.send(recipient, signed)

            # Validate response
            validated = self.validate_response(response)
            return validated
        else:
            raise BlockedException("Message blocked by governance")

    async def receive_message(self, message: dict):
        # Verify ES256 signature
        if not self.verify_signature_es256(message):
            raise SecurityException("Invalid ES256 signature")

        # Filter inbound
        filtered = self.message_filter.filter_inbound(message)

        # Create trace
        trace = CognitiveTrace(
            inputs={"message": filtered},
            meta={
                "protocol": "a2a",
                "sender": message["sender"]
            }
        )

        # Evaluate
        intervention = await self.steward.evaluate(trace)

        if intervention.is_allowed():
            return filtered
        else:
            await self.send_rejection(message["sender"], intervention.reason)
            return None

10. Conformance Requirements¶

10.1 General Requirements¶

A conformant implementation MUST: - Support at least one external protocol (MCP or A2A) - Implement all required governance injection points - Maintain protocol security requirements (TLS 1.3, ES256) - Meet performance targets for the supported protocols - Implement version negotiation

10.2 MCP Integration Requirements¶

For MCP integration, implementations MUST: - Intercept all tool calls before execution - Map MCP requests to Cognitive Traces correctly - Apply tool category-based governance - Evaluate tripwires before tool execution - Validate tool responses for sensitive data - Log all MCP interactions to ReflectionDB - Support all six intervention types (five primary levels: OK, Nudge, Escalate, Block, Halt, plus orthogonal Flag)

10.3 A2A Integration Requirements¶

For A2A integration, implementations MUST: - Verify agent identities using ACGP governance metadata - Filter messages based on ACL tiers - Detect and prevent collusion patterns using tripwires - Sign and verify all A2A messages with ES256 - Monitor collaborative tasks for boundary violations - Implement trust debt updates based on behavior

10.4 Security Requirements¶

All implementations MUST: - Use TLS 1.3 or higher for transport - Implement message signing with ES256 (standardized) - Detect and prevent replay attacks - Monitor for security anomalies - Maintain security audit logs

11. References¶

Normative References¶

ACGP-1000: Core Protocol Specification
ACGP-1001: Terminology and Definitions
ACGP-1002: Architecture Specification
ACGP-1003: Message Formats & Wire Protocol
ACGP-1005: ARS-CTQ-ACL Integration Framework
ACGP-1007: Security Considerations
RFC 2119: Key words for use in RFCs

Informative References¶

Model Context Protocol: https://modelcontext.org/spec
A2A Protocol Specification: https://a2a-protocol.org/spec
OAuth 2.0: RFC 6749
JSON Web Signature: RFC 7515

External Protocols¶

OpenAPI Specification: For REST API governance
GraphQL Schema: For GraphQL API governance
gRPC: For RPC governance
WebSocket: For real-time communication governance

End of ACGP-1008