MCP Server Security.

# MCP server configuration with per-connection authentication

# mcp_server_config.yaml
server:
  name: "company-tools"
  version: "1.0.0"

auth:
  type: "oauth2"
  token_endpoint: "https://auth.company.com/token"
  required_scopes:
    - "mcp:connect"

tool_permissions:
  # Per-tool authorization claims required
  filesystem_read:
    required_scope: "tools:filesystem:read"
  filesystem_write:
    required_scope: "tools:filesystem:write"
  network_request:
    required_scope: "tools:network:external"
    require_allowlist: true

  # Default: deny unlisted tools
  default_policy: "DENY"

# SupraWall wrapping an MCP-enabled agent

from suprawall import SupraWall
from anthropic import Anthropic

# Initialize SupraWall with MCP-specific policy
sw = SupraWall(
    api_key="sw_live_...",
    agent_id="document-processor",
    default_policy="DENY"
)

# Define tool-level policies for MCP tools
sw.apply_policies([
    # Allow only specific filesystem operations
    {"tool": "filesystem_read", "paths": ["/data/documents/*"], "action": "ALLOW"},
    {"tool": "filesystem_write", "action": "REQUIRE_APPROVAL"},

    # Block all network tools — this agent doesn't need external access
    {"tool": "network_*", "action": "DENY"},

    # Wildcard deny for anything not explicitly allowed
    {"tool": "*", "action": "DENY"},
])

# Create MCP client (standard Anthropic SDK)
client = Anthropic()

# SupraWall intercepts every tool call before MCP server receives it
@sw.intercept_mcp
def process_document(file_path: str):
    response = client.messages.create(
        model="claude-opus-4-5",
        tools=sw.get_allowed_tools(),  # Only allowed tools exposed
        messages=[{"role": "user", "content": f"Summarize {file_path}"}]
    )
    return response

# Every tool call in process_document is now policy-enforced
result = process_document("/data/documents/report.pdf")

# Complete MCP client with SupraWall enforcement

from suprawall import SupraWall, MCPPolicy
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Define MCP-specific policies
policies = MCPPolicy(
    agent_id="research-agent",
    allowed_tools=[
        "web_fetch",          # Allowed: web browsing
        "filesystem_read",    # Allowed: read documents
    ],
    blocked_tools=[
        "filesystem_write",   # Blocked: no file writes
        "email_*",            # Blocked: no email access
        "database_*",         # Blocked: no DB access
        "shell_exec",         # Blocked: no shell
    ],
    require_approval=[
        "network_post",       # Outbound POST requires human approval
    ],
    loop_detection_threshold=5,  # Block after 5 identical calls
    max_calls_per_session=200,   # Hard cap per session
)

sw = SupraWall(api_key="sw_live_...", policy=policies)

async def run_research_agent(query: str):
    server_params = StdioServerParameters(
        command="mcp-server-tools",
        args=["--config", "tools.yaml"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # SupraWall wraps the session — all tool calls intercepted
            protected_session = sw.wrap_mcp_session(session)

            # Agent uses protected_session for all tool calls
            # ALLOW -> call forwarded to MCP server
            # DENY  -> GuardrailError raised, call never reaches server
            # REQUIRE_APPROVAL -> paused, notification sent
            result = await agent.run(
                query=query,
                mcp_session=protected_session
            )

            return result