Security in MCP

Authentication

Verify server identity:

# Server presents certificate
server = MCPServer(
    cert_file="server.crt",
    key_file="server.key"
)

# Client verifies
client = MCPClient(
    server_url="https://...",
    verify_cert=True,
    trusted_certs=["server.crt"]
)

Authorization

Control what clients can access:

@server.tool("delete_user")
@require_permission("admin")
def delete_user(user_id: int):
    # Only admin can call this
    db.delete_user(user_id)

@server.tool("read_public_data")
@require_permission("user")
def read_data():
    # Any authenticated user
    return public_data

Input Validation

Prevent injection attacks:

@server.tool("query_db")
def query(sql: str):
    # BAD: Direct SQL
    # result = db.execute(sql)
    
    # GOOD: Parameterized query
    result = db.execute(
        "SELECT * FROM users WHERE id = ?",
        [sql]  # Treated as parameter, not code
    )
    return result

Output Sanitization

Don't leak sensitive data:

@server.tool("get_user")
def get_user(user_id: int):
    user = db.get(user_id)
    
    # BAD: Return all fields
    # return user  # Includes password_hash!
    
    # GOOD: Return only safe fields
    return {
        "id": user.id,
        "name": user.name,
        "email": user.email
        # password_hash NOT included
    }

Rate Limiting

Prevent DoS attacks:

from mcp.decorators import rate_limit

@server.tool("search")
@rate_limit(calls=100, period=60)  # 100 calls per minute
def search(query: str):
    return search_engine.search(query)

Sandboxing

Isolate untrusted code:

@server.tool("execute_code")
def run_code(code: str):
    # BAD: Direct execution
    # exec(code)  # Dangerous!
    
    # GOOD: Sandboxed environment
    sandbox = Sandbox(
        allowed_imports=["math", "random"],
        timeout=5,
        memory_limit="100MB"
    )
    result = sandbox.execute(code)
    return result

Logging & Auditing

Track all tool usage:

@server.tool("sensitive_operation")
def operation(user_id: int):
    logger.audit(
        event="tool_called",
        tool_name="sensitive_operation",
        user_id=user_id,
        timestamp=now(),
        source_ip=request.ip
    )
    return result