Adding AAP to MCP Tools

This guide shows how to extend MCP (Model Context Protocol) tools and servers with AAP alignment properties, enabling alignment verification for tool invocations.

Overview

MCP defines a protocol for exposing tools to language models. AAP extends MCP servers with alignment metadata that declares:

Who the server/tools serve (principal relationship)
What values guide tool behavior (declared values)
Which tools are bounded vs. require escalation (autonomy envelope)
How invocations are audited (trace commitment)

This extension enables clients to verify alignment before invoking tools, and produces AP-Traces for tool invocation auditing.

Prerequisites

pip install agent-alignment-protocol

MCP vs A2A: Key differences

Aspect	A2A	MCP
Scope	Agent capabilities	Tool invocations
Granularity	Agent-level cards	Server + tool-level alignment
Discovery	Agent Card endpoint	Server manifest + tool list
Actions	Skills	Tools

MCP alignment operates at two levels:

Server-level: Default alignment for all tools in the server
Tool-level: Override alignment for specific tools

Step 1: Understand your current MCP server

A standard MCP server exposes tools with JSON schemas:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="filesystem",
    instructions="File system operations for reading and writing files."
)

@mcp.tool()
def read_file(path: str) -> str:
    """Read contents of a file.

    Args:
        path: Path to the file to read

    Returns:
        File contents as string
    """
    with open(path) as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str) -> dict:
    """Write content to a file.

    Args:
        path: Path to the file to write
        content: Content to write

    Returns:
        Status dict with bytes written
    """
    with open(path, 'w') as f:
        bytes_written = f.write(content)
    return {"status": "success", "bytes_written": bytes_written}

@mcp.tool()
def delete_file(path: str) -> dict:
    """Delete a file from the filesystem.

    Args:
        path: Path to the file to delete

    Returns:
        Status dict
    """
    import os
    os.remove(path)
    return {"status": "deleted", "path": path}

This tells clients what tools are available, but not:

Which tools are safe to invoke autonomously
Which tools require user approval
Which paths/operations are forbidden
What values guide tool behavior

Step 2: Add server-level alignment

Create an alignment card for your MCP server:

from mcp.server.fastmcp import FastMCP
from aap import AlignmentCard

# Define server alignment
SERVER_ALIGNMENT = AlignmentCard(
    aap_version="1.0.0",
    card_id="ac-filesystem-server-001",
    agent_id="mcp-filesystem",
    issued_at="2026-01-31T12:00:00Z",

    principal={
        "type": "human",
        "relationship": "delegated_authority"
    },

    values={
        "declared": ["user_control", "transparency", "minimal_data"],
        "conflicts_with": ["data_exfiltration", "unauthorized_access"]
    },

    autonomy_envelope={
        "bounded_actions": ["read_file"],
        "escalation_triggers": [
            {
                "condition": "tool == 'write_file'",
                "action": "escalate",
                "reason": "Write operations require user approval"
            }
        ],
        "forbidden_actions": ["delete_file"]
    },

    audit_commitment={
        "trace_format": "ap-trace-v1",
        "retention_days": 30,
        "queryable": True,
        "query_endpoint": "mcp://filesystem/alignment/traces"
    }
)

mcp = FastMCP(
    name="filesystem",
    instructions="File system operations with alignment verification."
)

Key mapping: MCP tools to AAP actions

MCP Tool	AAP Treatment	Rationale
`read_file`	`bounded_actions`	Read-only, low risk
`write_file`	`escalation_triggers`	Modifies state, needs approval
`delete_file`	`forbidden_actions`	Destructive, never autonomous

Step 3: Expose alignment card

MCP servers SHOULD expose their alignment card via a resource:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="filesystem")

@mcp.resource("alignment://card")
def get_alignment_card() -> str:
    """Return the server's alignment card."""
    return SERVER_ALIGNMENT.model_dump_json(indent=2)

Alternatively, include alignment in server instructions:

mcp = FastMCP(
    name="filesystem",
    instructions=f"""File system operations with AAP alignment.

## Alignment Card

{SERVER_ALIGNMENT.model_dump_json(indent=2)}

## Tool Boundaries

- **Bounded (autonomous)**: read_file
- **Escalate (needs approval)**: write_file
- **Forbidden (never)**: delete_file
"""
)

Step 4: Generate AP-Traces for tool invocations

Wrap tool implementations to produce AP-Traces:

from aap import APTrace, Action, Decision, Escalation
from datetime import datetime, timezone
import uuid
import functools

def traced_tool(tool_name: str, category: str):
    """Decorator to generate AP-Traces for tool invocations."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            trace_id = f"tr-{uuid.uuid4().hex[:12]}"
            timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

            # Check if escalation is required
            escalation_required = category == "escalate"

            if category == "forbidden":
                # Log attempt but don't execute
                trace = APTrace(
                    trace_id=trace_id,
                    agent_id="mcp-filesystem",
                    card_id=SERVER_ALIGNMENT.card_id,
                    timestamp=timestamp,
                    action=Action(
                        type="tool_invocation",
                        name=tool_name,
                        category="forbidden",
                    ),
                    decision=Decision(
                        alternatives_considered=[],
                        selected=None,
                        selection_reasoning="Forbidden action blocked",
                        values_applied=["user_control"],
                    ),
                    escalation=Escalation(
                        evaluated=True,
                        triggers_checked=[{"trigger": "forbidden_action", "matched": True}],
                        required=True,
                        reason="Action is in forbidden_actions list",
                    ),
                )
                store_trace(trace)
                raise PermissionError(f"Forbidden action: {tool_name}")

            # Execute the tool
            result = func(*args, **kwargs)

            # Build trace
            trace = APTrace(
                trace_id=trace_id,
                agent_id="mcp-filesystem",
                card_id=SERVER_ALIGNMENT.card_id,
                timestamp=timestamp,
                action=Action(
                    type="tool_invocation",
                    name=tool_name,
                    category=category,
                ),
                decision=Decision(
                    alternatives_considered=[
                        {
                            "option_id": "execute",
                            "description": f"Execute {tool_name}",
                            "score": 1.0,
                            "flags": [],
                        }
                    ],
                    selected="execute",
                    selection_reasoning=f"Tool {tool_name} within autonomy envelope",
                    values_applied=SERVER_ALIGNMENT.values["declared"],
                ),
                escalation=Escalation(
                    evaluated=True,
                    triggers_checked=[
                        {"trigger": f"tool == '{tool_name}'", "matched": escalation_required}
                    ],
                    required=escalation_required,
                    reason="Within bounded actions" if not escalation_required else "Requires approval",
                ),
                context={
                    "tool_args": kwargs,
                    "result_summary": str(result)[:100] if result else None,
                },
            )

            store_trace(trace)
            return result
        return wrapper
    return decorator

# Apply to tools
@mcp.tool()
@traced_tool("read_file", "bounded")
def read_file(path: str) -> str:
    """Read contents of a file."""
    with open(path) as f:
        return f.read()

@mcp.tool()
@traced_tool("write_file", "escalate")
def write_file(path: str, content: str) -> dict:
    """Write content to a file (requires approval)."""
    with open(path, 'w') as f:
        bytes_written = f.write(content)
    return {"status": "success", "bytes_written": bytes_written}

@mcp.tool()
@traced_tool("delete_file", "forbidden")
def delete_file(path: str) -> dict:
    """Delete a file (forbidden action)."""
    import os
    os.remove(path)
    return {"status": "deleted"}

Step 5: Tool-level alignment overrides

For servers with many tools, specify per-tool alignment:

TOOL_ALIGNMENT = {
    "read_file": {
        "category": "bounded",
        "values": ["transparency", "minimal_data"],
        "conditions": [
            {"field": "path", "pattern": "^/home/", "action": "allow"},
            {"field": "path", "pattern": "^/etc/passwd", "action": "forbid"},
        ]
    },
    "write_file": {
        "category": "escalate",
        "values": ["user_control"],
        "escalation_reason": "Write operations modify system state",
    },
    "delete_file": {
        "category": "forbidden",
        "values": ["harm_prevention"],
        "forbidden_reason": "Destructive operations not permitted",
    },
    "list_directory": {
        "category": "bounded",
        "values": ["transparency"],
    },
}

def get_tool_category(tool_name: str) -> str:
    """Get alignment category for a tool."""
    return TOOL_ALIGNMENT.get(tool_name, {}).get("category", "escalate")

Step 6: Client-side verification

Clients invoking MCP tools can verify alignment before invocation:

from aap import verify_trace, check_coherence

class AlignedMCPClient:
    """MCP client with alignment verification."""

    def __init__(self, server_alignment: dict, my_alignment: dict):
        self.server_alignment = server_alignment
        self.my_alignment = my_alignment

    async def invoke_tool(self, tool_name: str, arguments: dict):
        """Invoke a tool with alignment checks."""

        # 1. Check value coherence with server
        coherence = check_coherence(self.my_alignment, self.server_alignment)
        if not coherence.proceed:
            raise ValueError(f"Value conflict with server: {coherence.value_alignment.conflicts}")

        # 2. Check if tool is within our autonomy envelope
        server_envelope = self.server_alignment.get("autonomy_envelope", {})
        bounded = server_envelope.get("bounded_actions", [])
        forbidden = server_envelope.get("forbidden_actions", [])

        if tool_name in forbidden:
            raise PermissionError(f"Tool {tool_name} is forbidden by server alignment")

        if tool_name not in bounded:
            # Requires escalation - check with principal
            approved = await self.request_approval(tool_name, arguments)
            if not approved:
                raise PermissionError(f"Tool {tool_name} requires approval (denied)")

        # 3. Invoke the tool
        result = await self._invoke(tool_name, arguments)

        # 4. Verify the trace (if server provides one)
        if hasattr(result, 'trace'):
            verification = verify_trace(result.trace, self.server_alignment)
            if not verification.verified:
                raise ValueError(f"Trace verification failed: {verification.violations}")

        return result

Complete example: Aligned MCP server

"""Filesystem MCP server with AAP alignment."""
from mcp.server.fastmcp import FastMCP
from aap import AlignmentCard, APTrace, Action, Decision, Escalation, verify_trace
from datetime import datetime, timezone
import uuid
import json
import os

# --- Alignment Configuration ---

SERVER_ALIGNMENT = AlignmentCard(
    aap_version="1.0.0",
    card_id="ac-filesystem-001",
    agent_id="mcp-filesystem",
    issued_at="2026-01-31T12:00:00Z",
    principal={"type": "human", "relationship": "delegated_authority"},
    values={
        "declared": ["user_control", "transparency", "harm_prevention"],
        "conflicts_with": ["data_exfiltration", "unauthorized_access"],
    },
    autonomy_envelope={
        "bounded_actions": ["read_file", "list_directory", "file_info"],
        "escalation_triggers": [
            {
                "condition": "tool in ['write_file', 'append_file']",
                "action": "escalate",
                "reason": "Write operations require approval"
            }
        ],
        "forbidden_actions": ["delete_file", "execute_command"],
    },
    audit_commitment={
        "trace_format": "ap-trace-v1",
        "retention_days": 90,
        "queryable": True,
        "query_endpoint": "mcp://filesystem/alignment/traces",
    },
)

# --- Trace Storage ---

TRACES: list[dict] = []

def store_trace(trace: APTrace):
    """Store trace for auditing."""
    TRACES.append(trace.model_dump(mode="json"))

# --- MCP Server ---

mcp = FastMCP(
    name="filesystem",
    instructions=f"""Filesystem operations with AAP alignment.

Alignment Card ID: {SERVER_ALIGNMENT.card_id}
Values: {', '.join(SERVER_ALIGNMENT.values['declared'])}

Tool Boundaries:
- Bounded (autonomous): read_file, list_directory, file_info
- Escalate (needs approval): write_file, append_file
- Forbidden (blocked): delete_file, execute_command
"""
)

# --- Alignment Resource ---

@mcp.resource("alignment://card")
def alignment_card() -> str:
    """Get server alignment card."""
    return SERVER_ALIGNMENT.model_dump_json(indent=2)

@mcp.resource("alignment://traces")
def alignment_traces() -> str:
    """Get recent AP-Traces."""
    return json.dumps(TRACES[-100:], indent=2)

# --- Tools with Tracing ---

@mcp.tool()
def read_file(path: str) -> str:
    """Read contents of a file (bounded action).

    Args:
        path: Path to the file to read

    Returns:
        File contents
    """
    trace_id = f"tr-{uuid.uuid4().hex[:12]}"
    timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    try:
        with open(path) as f:
            content = f.read()

        trace = APTrace(
            trace_id=trace_id,
            agent_id="mcp-filesystem",
            card_id=SERVER_ALIGNMENT.card_id,
            timestamp=timestamp,
            action=Action(type="tool_invocation", name="read_file", category="bounded"),
            decision=Decision(
                alternatives_considered=[{"option_id": "read", "description": f"Read {path}", "score": 1.0, "flags": []}],
                selected="read",
                selection_reasoning="Read-only operation within autonomy envelope",
                values_applied=["transparency"],
            ),
            escalation=Escalation(
                evaluated=True,
                triggers_checked=[{"trigger": "tool == 'read_file'", "matched": False}],
                required=False,
                reason="Bounded action",
            ),
            context={"path": path, "bytes_read": len(content)},
        )
        store_trace(trace)
        return content

    except Exception as e:
        # Trace the failure
        trace = APTrace(
            trace_id=trace_id,
            agent_id="mcp-filesystem",
            card_id=SERVER_ALIGNMENT.card_id,
            timestamp=timestamp,
            action=Action(type="tool_invocation", name="read_file", category="bounded"),
            decision=Decision(
                alternatives_considered=[],
                selected=None,
                selection_reasoning=f"Operation failed: {e}",
                values_applied=["transparency"],
            ),
            escalation=Escalation(evaluated=True, triggers_checked=[], required=False, reason="Failed"),
        )
        store_trace(trace)
        raise

@mcp.tool()
def write_file(path: str, content: str, approved: bool = False) -> dict:
    """Write content to a file (requires escalation).

    Args:
        path: Path to the file to write
        content: Content to write
        approved: Whether this write was explicitly approved by principal

    Returns:
        Status with bytes written
    """
    trace_id = f"tr-{uuid.uuid4().hex[:12]}"
    timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    if not approved:
        trace = APTrace(
            trace_id=trace_id,
            agent_id="mcp-filesystem",
            card_id=SERVER_ALIGNMENT.card_id,
            timestamp=timestamp,
            action=Action(type="tool_invocation", name="write_file", category="escalate"),
            decision=Decision(
                alternatives_considered=[],
                selected=None,
                selection_reasoning="Write operation requires explicit approval",
                values_applied=["user_control"],
            ),
            escalation=Escalation(
                evaluated=True,
                triggers_checked=[{"trigger": "tool == 'write_file'", "matched": True}],
                required=True,
                reason="Write operations require approval",
            ),
        )
        store_trace(trace)
        raise PermissionError("write_file requires approved=True (escalation)")

    with open(path, 'w') as f:
        bytes_written = f.write(content)

    trace = APTrace(
        trace_id=trace_id,
        agent_id="mcp-filesystem",
        card_id=SERVER_ALIGNMENT.card_id,
        timestamp=timestamp,
        action=Action(type="tool_invocation", name="write_file", category="escalate"),
        decision=Decision(
            alternatives_considered=[{"option_id": "write", "description": f"Write to {path}", "score": 1.0, "flags": ["approved"]}],
            selected="write",
            selection_reasoning="Write approved by principal",
            values_applied=["user_control", "transparency"],
        ),
        escalation=Escalation(
            evaluated=True,
            triggers_checked=[{"trigger": "tool == 'write_file'", "matched": True}],
            required=True,
            principal_response="approved",
            reason="Write operations require approval",
        ),
        context={"path": path, "bytes_written": bytes_written},
    )
    store_trace(trace)

    return {"status": "success", "bytes_written": bytes_written}

@mcp.tool()
def delete_file(path: str) -> dict:
    """Delete a file (forbidden action - always blocked).

    Args:
        path: Path to the file to delete

    Returns:
        Never returns - always raises
    """
    trace_id = f"tr-{uuid.uuid4().hex[:12]}"
    timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    trace = APTrace(
        trace_id=trace_id,
        agent_id="mcp-filesystem",
        card_id=SERVER_ALIGNMENT.card_id,
        timestamp=timestamp,
        action=Action(type="tool_invocation", name="delete_file", category="forbidden"),
        decision=Decision(
            alternatives_considered=[],
            selected=None,
            selection_reasoning="Forbidden action blocked",
            values_applied=["harm_prevention", "user_control"],
        ),
        escalation=Escalation(
            evaluated=True,
            triggers_checked=[{"trigger": "forbidden_action", "matched": True}],
            required=True,
            reason="delete_file is in forbidden_actions",
        ),
    )
    store_trace(trace)

    raise PermissionError("delete_file is a forbidden action")

# --- Verification Endpoint ---

@mcp.tool()
def verify_recent_traces(limit: int = 10) -> dict:
    """Verify recent traces against the alignment card.

    Args:
        limit: Number of recent traces to verify

    Returns:
        Verification summary
    """
    recent = TRACES[-limit:]
    results = []

    for trace_dict in recent:
        trace = APTrace(**trace_dict)
        result = verify_trace(trace, SERVER_ALIGNMENT)
        results.append({
            "trace_id": trace.trace_id,
            "passed": result.verified,
            "violations": [v.description for v in result.violations] if result.violations else [],
        })

    passed = sum(1 for r in results if r["passed"])

    return {
        "total": len(results),
        "passed": passed,
        "failed": len(results) - passed,
        "details": results,
    }

if __name__ == "__main__":
    mcp.run()

MCP configuration with Alignment

Update your .mcp.json to indicate alignment support:

{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["-m", "filesystem_server"],
      "env": {
        "AAP_TRACE_ENABLED": "true",
        "AAP_TRACE_PATH": "/var/log/aap/filesystem/"
      },
      "metadata": {
        "alignment": {
          "supported": true,
          "card_resource": "alignment://card",
          "traces_resource": "alignment://traces"
        }
      }
    }
  }
}

Migration checklist

Handling non-AAP clients

MCP servers with AAP should still work with clients that don’t support alignment:

@mcp.tool()
def write_file(path: str, content: str, approved: bool = False) -> dict:
    """Write content to a file.

    Args:
        path: Path to write
        content: Content to write
        approved: AAP approval flag (non-AAP clients can omit)
    """
    # Non-AAP clients won't pass approved=True
    # Server policy: require approval for writes
    if not approved:
        return {
            "status": "escalation_required",
            "message": "This operation requires approval. Call with approved=True after user confirms.",
            "aap_info": {
                "action_category": "escalate",
                "reason": "Write operations require explicit approval",
                "card_id": SERVER_ALIGNMENT.card_id,
            }
        }

    # Proceed with approved write
    ...

Standard value identifiers

Use these standard identifiers where applicable:

Identifier	Description
`user_control`	Respect user autonomy and consent
`transparency`	Disclose operations and reasoning
`minimal_data`	Access only necessary data
`harm_prevention`	Avoid destructive operations
`honesty`	Do not deceive or mislead
`privacy`	Protect personal information
`principal_benefit`	Prioritize principal’s interests

Custom values MUST be defined in the alignment card’s definitions block.

Next steps

quickstart — Core AAP concepts and API
specification — Full protocol specification
A2A integration — Adding AAP to A2A agents
examples — Working example code

Questions? See the specification or check the examples.

Protocols

Agent Alignment Protocol

Agent Integrity Protocol

Adding AAP to MCP Tools

Overview

Prerequisites

MCP vs A2A: Key differences

Step 1: Understand your current MCP server

Step 2: Add server-level alignment

Key mapping: MCP tools to AAP actions

Step 3: Expose alignment card

Step 4: Generate AP-Traces for tool invocations

Step 5: Tool-level alignment overrides

Step 6: Client-side verification

Complete example: Aligned MCP server

MCP configuration with Alignment

Migration checklist

Handling non-AAP clients

Standard value identifiers

Next steps

​Overview

​Prerequisites

​MCP vs A2A: Key differences

​Step 1: Understand your current MCP server

​Step 2: Add server-level alignment

​Key mapping: MCP tools to AAP actions

​Step 3: Expose alignment card

​Step 4: Generate AP-Traces for tool invocations

​Step 5: Tool-level alignment overrides

​Step 6: Client-side verification

​Complete example: Aligned MCP server

​MCP configuration with Alignment

​Migration checklist

​Handling non-AAP clients

​Standard value identifiers

​Next steps

Overview

Prerequisites

MCP vs A2A: Key differences

Step 1: Understand your current MCP server

Step 2: Add server-level alignment

Key mapping: MCP tools to AAP actions

Step 3: Expose alignment card

Step 4: Generate AP-Traces for tool invocations

Step 5: Tool-level alignment overrides

Step 6: Client-side verification

Complete example: Aligned MCP server

MCP configuration with Alignment

Migration checklist

Handling non-AAP clients

Standard value identifiers

Next steps