> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mnemom.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Adding AAP to MCP Tools

> How to extend MCP tools and servers with AAP alignment properties for alignment verification of tool invocations

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

```bash theme={null}
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:

1. **Server-level**: Default alignment for all tools in the server
2. **Tool-level**: Override alignment for specific tools

## Step 1: Understand your current MCP server

A standard MCP server exposes tools with JSON schemas:

```python theme={null}
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:

```python theme={null}
from mcp.server.fastmcp import FastMCP
from aap import AlignmentCard

# Define server alignment
SERVER_ALIGNMENT = AlignmentCard(
    card_version="unified/2026-04-26",
    autonomy_mode="observe",
    integrity_mode="observe",
    card_id="ac-filesystem-server-001",
    agent_id="mcp-filesystem",
    issued_at="2026-01-31T12:00:00Z",

    principal={
        "type": "human",
        "identifier": "mcp-filesystem-owner",
        "relationship": "delegated_authority"
    },

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

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

    audit={
        "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:

```python theme={null}
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:

```python theme={null}
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:

`Action.type` is one of `recommend | execute | escalate | deny` (there is no `tool_invocation` value), and `Action.category` is one of `bounded | escalation_trigger | forbidden` (not `"escalate"`). `Decision.alternatives_considered` requires at least one entry, and `selected` MUST be one of those entries' `option_id` -- it cannot be `None` or reference an empty list. `Decision.values_applied` must be a plain list of strings, so read `SERVER_ALIGNMENT.values.declared` (an attribute on the `Values` model), not `SERVER_ALIGNMENT.values["declared"]`:

```python theme={null}
from aap import APTrace, Action, Decision, Escalation
from datetime import datetime, timezone
import uuid
import functools

# Maps the decorator's simple category to the real ActionCategory enum
_CATEGORY_MAP = {
    "bounded": "bounded",
    "escalate": "escalation_trigger",
    "forbidden": "forbidden",
}

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")
            action_category = _CATEGORY_MAP[category]
            escalation_required = category == "escalate"

            if category == "forbidden":
                # Log the attempt but don't execute. alternatives_considered
                # still needs >= 1 entry, and selected must reference it.
                trace = APTrace(
                    trace_id=trace_id,
                    agent_id="mcp-filesystem",
                    card_id=SERVER_ALIGNMENT.card_id,
                    timestamp=timestamp,
                    action=Action(
                        type="deny",
                        name=tool_name,
                        category="forbidden",
                    ),
                    decision=Decision(
                        alternatives_considered=[
                            {"option_id": "deny", "description": f"Deny {tool_name}"},
                        ],
                        selected="deny",
                        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="execute",
                    name=tool_name,
                    category=action_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={
                    "metadata": {
                        "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:

```python theme={null}
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:

```python theme={null}
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") or 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

```python theme={null}
"""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(
    card_version="unified/2026-04-26",
    autonomy_mode="observe",
    integrity_mode="observe",
    card_id="ac-filesystem-001",
    agent_id="mcp-filesystem",
    issued_at="2026-01-31T12:00:00Z",
    principal={"type": "human", "identifier": "mcp-filesystem-owner", "relationship": "delegated_authority"},
    values={
        "declared": ["user_control", "transparency", "harm_prevention"],
        "conflicts_with": ["data_exfiltration", "unauthorized_access"],
    },
    autonomy={
        "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={
        "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="execute", 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={"metadata": {"path": path, "bytes_read": len(content)}},
        )
        store_trace(trace)
        return content

    except Exception as e:
        # Trace the failure. alternatives_considered still needs >= 1 entry,
        # and selected must reference one of them -- neither can be empty/None.
        trace = APTrace(
            trace_id=trace_id,
            agent_id="mcp-filesystem",
            card_id=SERVER_ALIGNMENT.card_id,
            timestamp=timestamp,
            action=Action(type="deny", name="read_file", category="bounded"),
            decision=Decision(
                alternatives_considered=[{"option_id": "failed", "description": "Read attempt failed"}],
                selected="failed",
                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="escalate", name="write_file", category="escalation_trigger"),
            decision=Decision(
                alternatives_considered=[{"option_id": "blocked", "description": "Write blocked pending approval"}],
                selected="blocked",
                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="execute", name="write_file", category="escalation_trigger"),
        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 is a PrincipalResponse object, not a bare string
            principal_response={"decision": "approved", "timestamp": timestamp},
            reason="Write operations require approval",
        ),
        context={"metadata": {"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="deny", name="delete_file", category="forbidden"),
        decision=Decision(
            alternatives_considered=[{"option_id": "deny", "description": "Delete blocked"}],
            selected="deny",
            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:

```json theme={null}
{
  "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

* [ ] Audit your current MCP tools
* [ ] Classify tools: bounded, escalate, or forbidden
* [ ] Create server-level alignment card
* [ ] Define tool-level overrides if needed
* [ ] Add alignment card resource (`alignment://card`)
* [ ] Implement AP-Trace generation for tool invocations
* [ ] Add trace storage/retrieval resource
* [ ] Test with `verify_trace()` before deployment
* [ ] Update `.mcp.json` with alignment metadata
* [ ] Document alignment in server instructions
* [ ] Handle non-AAP clients gracefully

## Handling non-AAP clients

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

```python theme={null}
@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](/protocols/aap/quickstart)** -- Core AAP concepts and API
* **[specification](/protocols/aap/specification)** -- Full protocol specification
* **[A2A integration](/protocols/aap/a2a-integration)** -- Adding AAP to A2A agents
* **[examples](https://github.com/mnemom/aap/tree/main/examples/mcp-integration/)** -- Working example code

***

*Questions? See the [specification](/protocols/aap/specification) or check the [examples](https://github.com/mnemom/aap/tree/main/examples).*


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.