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:

OWASP MCP Risk	Coverage	What crewai-mcps does
MCP-01: Tool Poisoning	Full	Pins tool definitions at discovery; blocks execution if definition changes
MCP-04: Tool Rug Pulls	Full	Hashes tool schemas on first contact; any mutation raises ToolIntegrityError
MCP-08: Logging Gaps	Full	Structured audit trail with sequence numbers, timestamps, and exportable JSON
MCP-10: Lack of Integrity	Full	Cryptographic signatures on canonical JSON payloads; tampered arguments fail verification
Replay Attacks	Full	Every call carries a unique nonce with configurable TTL; duplicates are rejected

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

Class	Purpose
SecureMCPTool	Main entry point. Wraps tools, signs calls, verifies responses
SecureToolProxy	Transparent proxy that intercepts tool execution
KeyManager	Manages cryptographic key pairs
NonceTracker	Generates and validates nonces with TTL
ToolPinner	Hashes and pins tool definitions
ResponseVerifier	Matches responses to requests with timing checks
AuditTrail	Thread-safe structured event log
install_crewai_hooks	Registers MCPS as global CrewAI tool hooks

Requirements

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

License

Apache 2.0. See LICENSE.