crewai-mcps

Cryptographic security layer for CrewAI MCP tool calls.

crewai-mcps adds message signing, nonce-based replay protection, and tool definition pinning to CrewAI's Model Context Protocol (MCP) integrations. It addresses specific risks from the OWASP MCP Top 10:

MCP-02 (permissions), MCP-06 (resource injection), MCP-07 (auth), and MCP-09 (resource abuse) are outside the scope of this package -- they require policy-layer enforcement. See ELIDA for behavioural-layer coverage of all 10.

Installation

pip install crewai-mcps

With CrewAI:

pip install 'crewai-mcps[crewai]'

Quick Start

Wrap MCP tools (recommended)

from crewai import Agent, Task, Crew
from crewai.mcp import MCPServerAdapter
from crewai_mcps import SecureMCPTool, AuditTrail

# Optional: audit trail for compliance
audit = AuditTrail()

# Zero config -- keys generated automatically
secure = SecureMCPTool(audit_trail=audit)

server_params = {
    "url": "http://localhost:8001/mcp",
    "transport": "streamable-http",
}

with MCPServerAdapter(server_params) as tools:
    # One line to add security
    secured_tools = secure.wrap_tools(tools)

    agent = Agent(
        role="Analyst",
        goal="Perform secure analysis",
        backstory="All tool calls are cryptographically signed.",
        tools=secured_tools,
    )

    task = Task(
        description="List available files.",
        expected_output="A file listing.",
        agent=agent,
    )

    crew = Crew(agents=[agent], tasks=[task])
    result = crew.kickoff()

# Review what happened
print(audit.summary())
audit.export_json("audit.json")

Global hooks (signs all tool calls)

from crewai_mcps import SecureMCPTool, install_crewai_hooks

secure = SecureMCPTool()
cleanup = install_crewai_hooks(secure)

# All tool calls are now signed automatically
# ... run your crew ...

cleanup()  # Remove hooks when done

Features

Cryptographic Signing

Every tool call is signed with an automatically generated key pair. The signature covers the canonical JSON of the tool name, arguments, nonce, and timestamp. Any tampering with the arguments invalidates the signature.

secure = SecureMCPTool()

# Sign a call manually
envelope = secure.sign_call("search", {"query": "test"})
# => {'nonce': '...', 'timestamp': '...', 'signature': '...', 'tool_name': 'search'}

Replay Protection

Each call carries a unique nonce. The NonceTracker maintains a time-windowed set of used nonces (default: 5 minutes). Duplicate nonces are rejected.

from crewai_mcps import NonceTracker

tracker = NonceTracker(max_age_seconds=300)
nonce = tracker.generate()

# First use: OK
tracker.check_and_consume(nonce)  # True

# Replay attempt: blocked
tracker.check_and_consume(nonce)  # False

Tool Pinning (Rug-Pull Detection)

When tools are discovered from an MCP server, their definitions (name, description, schema) are hashed and pinned. If the server changes a tool's definition between calls, execution is blocked with a ToolIntegrityError.

from crewai_mcps import ToolPinner

pinner = ToolPinner()
pinner.pin("write_file", {
    "name": "write_file",
    "description": "Write a file to disk",
    "schema": {"type": "object", "properties": {"path": {"type": "string"}}},
})

# Later: server changes the tool
pinner.verify("write_file", {
    "name": "write_file",
    "description": "Write a file to disk",
    "schema": {
        "type": "object",
        "properties": {
            "path": {"type": "string"},
            "exec_cmd": {"type": "string"},  # Injected!
        },
    },
})
# => False

Structured Audit Trail

Every signed call, response, error, and security event is recorded with monotonically increasing sequence numbers. Export the full trail for compliance.

from crewai_mcps import AuditTrail

audit = AuditTrail(max_events=10000)

# Events are recorded automatically when used with SecureMCPTool
secure = SecureMCPTool(audit_trail=audit)

# Query the trail
audit.get_events_for_tool("search")
audit.get_security_alerts()
audit.get_errors()

# Export
audit.export_json("trail.json")
print(audit.summary())

Response Verification

Standalone verifier for matching responses to signed requests with timing checks.

from crewai_mcps import ResponseVerifier

verifier = ResponseVerifier(max_response_age=30.0)
verifier.register_request("nonce-1", "search", {"q": "test"}, envelope)

# Later...
result = verifier.verify_response("nonce-1", "search", response)
if result.valid:
    print("Response verified")
else:
    print(f"Verification failed: {result.errors}")

Architecture

CrewAI Agent
    |
    v
SecureToolProxy (wraps each MCP tool)
    |
    +-- ToolPinner.verify()     # Check definition integrity
    +-- SecureMCPTool.sign()    # Sign with nonce + timestamp
    +-- original_tool.run()     # Execute the real tool
    +-- verify_response()       # Log and verify result
    +-- AuditTrail.log()        # Record everything

API Reference

Requirements

• Python 3.10+
• cryptography >= 41.0
• crewai >= 0.80.0 (optional, for hook integration)

License

Apache 2.0. See LICENSE.