pip install agent-alignment-protocol

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}

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

# Define server alignment
SERVER_ALIGNMENT = AlignmentCard(
    aap_version="0.1.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."
)

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)

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
"""
)

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"}

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")

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

"""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="0.1.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()

{
  "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"
        }
      }
    }
  }
}

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