Using Pipelock with CrewAI#

Pipelock wraps CrewAI's MCP servers as a stdio proxy, scanning every request and
response for credential leaks, prompt injection, and tool description poisoning.
This guide covers both the mcps= DSL and the MCPServerAdapter approach.

Quick Start#

# 1. Install pipelock
git clone --branch v3.5.0 --depth 1 https://github.com/luckyPipewrench/pipelock.git
make -C pipelock install

# 2. Generate a config (or copy a preset)
pipelock generate config --preset generic-agent > pipelock.yaml

# 3. Verify
pipelock version

from crewai import Agent
from crewai.mcp import MCPServerStdio

agent = Agent(
    role="Research Analyst",
    goal="Research and analyze information",
    backstory="Expert researcher with access to external tools",
    mcps=[
        MCPServerStdio(
            command="pipelock",
            args=[
                "mcp", "proxy",
                "--config", "pipelock.yaml",
                "--",
                "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"
            ],
        ),
    ],
)

That's it. Pipelock intercepts all MCP traffic between CrewAI and the filesystem
server, scanning in both directions.

How It Works#

CrewAI Agent <--> pipelock mcp proxy <--> MCP Server
  (client) (scan both ways) (subprocess)

Pipelock scans three things:

Outbound requests. Catches credentials leaking through tool arguments
(API keys, tokens, private key material).
Inbound responses. Catches prompt injection in tool results.
Tool descriptions. Catches poisoned tool definitions and mid-session
rug-pull changes.

Integration Patterns#

Pattern A: `mcps=` DSL (CrewAI >= 1.4.0)#

The mcps field on Agent accepts MCPServerStdio objects. Set command to
pipelock and put the real server command after -- in args:

from crewai import Agent, Task, Crew
from crewai.mcp import MCPServerStdio

filesystem_server = MCPServerStdio(
    command="pipelock",
    args=[
        "mcp", "proxy",
        "--config", "pipelock.yaml",
        "--",
        "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"
    ],
)

database_server = MCPServerStdio(
    command="pipelock",
    args=[
        "mcp", "proxy",
        "--config", "pipelock.yaml",
        "--",
        "python", "-m", "mcp_server_sqlite", "--db", "/data/app.db"
    ],
)

agent = Agent(
    role="Data Analyst",
    goal="Query data and generate reports",
    backstory="An analyst with access to files and databases",
    mcps=[filesystem_server, database_server],
)

task = Task(
    description="Read the sales data from /workspace/sales.csv and summarize it",
    expected_output="A summary of key sales metrics",
    agent=agent,
)

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

Each server gets its own Pipelock proxy instance. If one server returns a
poisoned response, Pipelock blocks it without affecting the other.

Pattern B: `MCPServerAdapter` (crewai-tools)#

The adapter approach uses a context manager and passes tools via the tools=
parameter:

from crewai import Agent
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters

server_params = StdioServerParameters(
    command="pipelock",
    args=[
        "mcp", "proxy",
        "--config", "pipelock.yaml",
        "--",
        "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"
    ],
)

with MCPServerAdapter(server_params) as tools:
    agent = Agent(
        role="File Processor",
        goal="Process files in the workspace",
        backstory="An agent with filesystem access",
        tools=tools,
    )
    # ... create tasks and crew

Pattern C: Multiple servers with mixed transports#

Wrap each stdio server individually. Remote servers (SSE/HTTP) connect directly
to the remote endpoint and are not covered by the stdio proxy:

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerSSE

agent = Agent(
    role="Multi-tool Agent",
    goal="Use local and remote tools",
    backstory="Agent with diverse tool access",
    mcps=[
        # Local server: wrap with pipelock stdio proxy
        MCPServerStdio(
            command="pipelock",
            args=["mcp", "proxy", "--config", "pipelock.yaml", "--",
                  "npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        ),
        # Direct remote server: NOT scanned by Pipelock.
        # To scan this path, expose the remote endpoint through:
        # pipelock mcp proxy --upstream https://api.example.com/mcp/sse
        MCPServerSSE(
            url="https://api.example.com/mcp/sse",
            headers={"Authorization": "Bearer token"},
        ),
    ],
)

Note: Direct MCPServerSSE / HTTP MCP connections go straight to the remote
endpoint and bypass Pipelock. To scan remote MCP traffic you have two options:

pipelock mcp proxy --upstream https://api.example.com/mcp/sse registers
Pipelock as a stdio MCP server that bridges to the remote endpoint. Use this
when the upstream needs no client-supplied HTTP headers, since the stdio
bridge does not have a transparent path for the client's Authorization or
other custom headers.
HTTP reverse proxy mode (pipelock run --mcp-listen ADDR --mcp-upstream URL)
preserves request headers through to the upstream. Use this when the upstream
requires per-request Authorization or other client-supplied headers, or
when the client needs an HTTP MCP URL.

For outbound HTTP traffic from your agent code (API calls, web fetches), route
those through pipelock run as a fetch proxy. See the
HTTP fetch proxy section below.

What Pipelock Catches#

CrewAI's MCP security documentation
identifies several threats. Here's what Pipelock detects at runtime:

Threat	CrewAI's Guidance	Pipelock's Detection
Credential exposure	"Use trusted servers"	DLP scanner catches 65 built-in credential patterns in tool arguments (API keys, tokens, private keys)
Prompt injection via tool metadata	"Only connect to trusted servers"	Tool description scanning detects hidden instructions, exfiltration directives, cross-tool manipulation
Tool behavior changes mid-session	Not addressed	Rug-pull detection tracks tool definition hashes per session, alerts on changes
DNS exfiltration via subdomains	Not addressed	Dot-collapse scanning and subdomain entropy analysis
Data exfiltration through responses	Not addressed	Response scanning with injection pattern matching

CrewAI's tool_filter (allowlist/blocklist by name) controls which tools the
agent sees. Pipelock scans what those tools actually say and do.

Docker Compose#

Network-isolated deployment where the agent container has no direct internet
access:

networks:
  pipelock-internal:
    internal: true
    driver: bridge
  pipelock-external:
    driver: bridge

services:
  pipelock:
    image: ghcr.io/luckypipewrench/pipelock:3.5.0
    networks:
      - pipelock-internal
      - pipelock-external
    command: ["run", "--listen", "0.0.0.0:8888", "--config", "/config/pipelock.yaml"]
    volumes:
      - ./pipelock.yaml:/config/pipelock.yaml:ro
    healthcheck:
      test: ["/pipelock", "healthcheck"]
      interval: 10s
      timeout: 3s
      start_period: 5s
      retries: 3

  crewai-agent:
    build: .
    networks:
      - pipelock-internal
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - HTTP_PROXY=http://pipelock:8888
      - HTTPS_PROXY=http://pipelock:8888
      - NO_PROXY=localhost,127.0.0.1
    depends_on:
      pipelock:
        condition: service_healthy

The agent container can only reach the pipelock service. The proxy
environment variables route HTTP libraries that honor them through Pipelock.
PIPELOCK_FETCH_URL is an application-defined helper value: it is useful only
when your code explicitly calls the /fetch endpoint and does not redirect a
CrewAI request by itself. MCP servers running as subprocesses inside the agent
container are wrapped with pipelock mcp proxy as shown above.

You can also generate this template with:

pipelock generate docker-compose --agent generic

HTTP Fetch Proxy#

For scanning HTTP traffic from CrewAI agents (web searches, API calls), run
Pipelock as a fetch proxy:

pipelock generate config --preset balanced > pipelock.yaml
pipelock run --config pipelock.yaml

Configure your agent to route HTTP requests through http://localhost:8888/fetch.
The endpoint returns JSON with content, blocked, and block_reason fields:

import requests

def fetch_through_pipelock(url: str) -> str:
    resp = requests.get(
        "http://localhost:8888/fetch",
        params={"url": url},
        headers={"X-Pipelock-Agent": "crewai-research"},
    )
    data = resp.json()
    if data.get("blocked"):
        raise RuntimeError(f"Pipelock blocked request: {data.get('block_reason')}")
    return data.get("content", "")

TLS Interception#

When using pipelock as an HTTP forward proxy (HTTPS_PROXY), CONNECT tunnels
are opaque by default: pipelock only sees the hostname, not the request body or
response content. Enabling TLS interception closes this gap by performing a MITM
on HTTPS connections, giving you full DLP on request bodies and response
injection detection through CONNECT tunnels.

To enable it:

Generate a CA and enable TLS interception (see the TLS Interception Guide)
Trust the CA in your Python environment:

export SSL_CERT_FILE=~/.pipelock/ca.pem
# Or for requests/httpx specifically:
export REQUESTS_CA_BUNDLE=~/.pipelock/ca.pem

MCP proxy mode (stdio wrapping) does not require TLS interception. It scans
traffic in both directions without certificates.

Choosing a Config#

Preset	Action	Best For
`generic-agent`	warn	New agent integrations (recommended starting point)
`balanced`	warn	General purpose, fetch proxy tuning
`claude-code`	block	Claude Code, unattended agents
`strict`	block	High-security, production

Start with generic-agent to log detections without blocking. Review the
logs, tune thresholds, then switch to strict for production.

Troubleshooting#

MCP server not starting#

Verify the command works without Pipelock first:

npx -y @modelcontextprotocol/server-filesystem /tmp

Then wrap it:

pipelock mcp proxy -- npx -y @modelcontextprotocol/server-filesystem /tmp

Seeing Pipelock output#

Pipelock logs to stderr. In CrewAI, stderr from stdio subprocesses may be
captured. To see real-time output during development:

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | \
  pipelock mcp proxy --config pipelock.yaml -- npx -y @modelcontextprotocol/server-filesystem /tmp

False positives#

Switch to warn mode to see what's being flagged without blocking:

response_scanning:
  action: warn
mcp_input_scanning:
  action: warn
mcp_tool_scanning:
  action: warn

Review stderr output, then tighten thresholds.

Config file not found#

Use absolute paths if relative paths don't resolve in your CrewAI project
structure:

MCPServerStdio(
    command="pipelock",
    args=["mcp", "proxy", "--config", "/etc/pipelock/config.yaml", "--",
          "npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
)

Using Pipelock with CrewAI#

Quick Start#

# 1. Install pipelock
git clone --branch v3.5.0 --depth 1 https://github.com/luckyPipewrench/pipelock.git
make -C pipelock install

# 2. Generate a config (or copy a preset)
pipelock generate config --preset generic-agent > pipelock.yaml

# 3. Verify
pipelock version

from crewai import Agent
from crewai.mcp import MCPServerStdio

agent = Agent(
    role="Research Analyst",
    goal="Research and analyze information",
    backstory="Expert researcher with access to external tools",
    mcps=[
        MCPServerStdio(
            command="pipelock",
            args=[
                "mcp", "proxy",
                "--config", "pipelock.yaml",
                "--",
                "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"
            ],
        ),
    ],
)

That's it. Pipelock intercepts all MCP traffic between CrewAI and the filesystem
server, scanning in both directions.

How It Works#

CrewAI Agent <--> pipelock mcp proxy <--> MCP Server
  (client) (scan both ways) (subprocess)

Pipelock scans three things:

Outbound requests. Catches credentials leaking through tool arguments
(API keys, tokens, private key material).
Inbound responses. Catches prompt injection in tool results.
Tool descriptions. Catches poisoned tool definitions and mid-session
rug-pull changes.

Integration Patterns#

Pattern A: `mcps=` DSL (CrewAI >= 1.4.0)#

The mcps field on Agent accepts MCPServerStdio objects. Set command to
pipelock and put the real server command after -- in args:

from crewai import Agent, Task, Crew
from crewai.mcp import MCPServerStdio

filesystem_server = MCPServerStdio(
    command="pipelock",
    args=[
        "mcp", "proxy",
        "--config", "pipelock.yaml",
        "--",
        "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"
    ],
)

database_server = MCPServerStdio(
    command="pipelock",
    args=[
        "mcp", "proxy",
        "--config", "pipelock.yaml",
        "--",
        "python", "-m", "mcp_server_sqlite", "--db", "/data/app.db"
    ],
)

agent = Agent(
    role="Data Analyst",
    goal="Query data and generate reports",
    backstory="An analyst with access to files and databases",
    mcps=[filesystem_server, database_server],
)

task = Task(
    description="Read the sales data from /workspace/sales.csv and summarize it",
    expected_output="A summary of key sales metrics",
    agent=agent,
)

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

Each server gets its own Pipelock proxy instance. If one server returns a
poisoned response, Pipelock blocks it without affecting the other.

Pattern B: `MCPServerAdapter` (crewai-tools)#

The adapter approach uses a context manager and passes tools via the tools=
parameter:

from crewai import Agent
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters

server_params = StdioServerParameters(
    command="pipelock",
    args=[
        "mcp", "proxy",
        "--config", "pipelock.yaml",
        "--",
        "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"
    ],
)

with MCPServerAdapter(server_params) as tools:
    agent = Agent(
        role="File Processor",
        goal="Process files in the workspace",
        backstory="An agent with filesystem access",
        tools=tools,
    )
    # ... create tasks and crew

Pattern C: Multiple servers with mixed transports#

Wrap each stdio server individually. Remote servers (SSE/HTTP) connect directly
to the remote endpoint and are not covered by the stdio proxy:

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerSSE

agent = Agent(
    role="Multi-tool Agent",
    goal="Use local and remote tools",
    backstory="Agent with diverse tool access",
    mcps=[
        # Local server: wrap with pipelock stdio proxy
        MCPServerStdio(
            command="pipelock",
            args=["mcp", "proxy", "--config", "pipelock.yaml", "--",
                  "npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        ),
        # Direct remote server: NOT scanned by Pipelock.
        # To scan this path, expose the remote endpoint through:
        # pipelock mcp proxy --upstream https://api.example.com/mcp/sse
        MCPServerSSE(
            url="https://api.example.com/mcp/sse",
            headers={"Authorization": "Bearer token"},
        ),
    ],
)

Note: Direct MCPServerSSE / HTTP MCP connections go straight to the remote
endpoint and bypass Pipelock. To scan remote MCP traffic you have two options:

pipelock mcp proxy --upstream https://api.example.com/mcp/sse registers
Pipelock as a stdio MCP server that bridges to the remote endpoint. Use this
when the upstream needs no client-supplied HTTP headers, since the stdio
bridge does not have a transparent path for the client's Authorization or
other custom headers.
HTTP reverse proxy mode (pipelock run --mcp-listen ADDR --mcp-upstream URL)
preserves request headers through to the upstream. Use this when the upstream
requires per-request Authorization or other client-supplied headers, or
when the client needs an HTTP MCP URL.

For outbound HTTP traffic from your agent code (API calls, web fetches), route
those through pipelock run as a fetch proxy. See the
HTTP fetch proxy section below.

What Pipelock Catches#

CrewAI's MCP security documentation
identifies several threats. Here's what Pipelock detects at runtime:

Threat	CrewAI's Guidance	Pipelock's Detection
Credential exposure	"Use trusted servers"	DLP scanner catches 65 built-in credential patterns in tool arguments (API keys, tokens, private keys)
Prompt injection via tool metadata	"Only connect to trusted servers"	Tool description scanning detects hidden instructions, exfiltration directives, cross-tool manipulation
Tool behavior changes mid-session	Not addressed	Rug-pull detection tracks tool definition hashes per session, alerts on changes
DNS exfiltration via subdomains	Not addressed	Dot-collapse scanning and subdomain entropy analysis
Data exfiltration through responses	Not addressed	Response scanning with injection pattern matching

CrewAI's tool_filter (allowlist/blocklist by name) controls which tools the
agent sees. Pipelock scans what those tools actually say and do.

Docker Compose#

Network-isolated deployment where the agent container has no direct internet
access:

networks:
  pipelock-internal:
    internal: true
    driver: bridge
  pipelock-external:
    driver: bridge

services:
  pipelock:
    image: ghcr.io/luckypipewrench/pipelock:3.5.0
    networks:
      - pipelock-internal
      - pipelock-external
    command: ["run", "--listen", "0.0.0.0:8888", "--config", "/config/pipelock.yaml"]
    volumes:
      - ./pipelock.yaml:/config/pipelock.yaml:ro
    healthcheck:
      test: ["/pipelock", "healthcheck"]
      interval: 10s
      timeout: 3s
      start_period: 5s
      retries: 3

  crewai-agent:
    build: .
    networks:
      - pipelock-internal
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - HTTP_PROXY=http://pipelock:8888
      - HTTPS_PROXY=http://pipelock:8888
      - NO_PROXY=localhost,127.0.0.1
    depends_on:
      pipelock:
        condition: service_healthy

You can also generate this template with:

pipelock generate docker-compose --agent generic

HTTP Fetch Proxy#

For scanning HTTP traffic from CrewAI agents (web searches, API calls), run
Pipelock as a fetch proxy:

pipelock generate config --preset balanced > pipelock.yaml
pipelock run --config pipelock.yaml

Configure your agent to route HTTP requests through http://localhost:8888/fetch.
The endpoint returns JSON with content, blocked, and block_reason fields:

import requests

def fetch_through_pipelock(url: str) -> str:
    resp = requests.get(
        "http://localhost:8888/fetch",
        params={"url": url},
        headers={"X-Pipelock-Agent": "crewai-research"},
    )
    data = resp.json()
    if data.get("blocked"):
        raise RuntimeError(f"Pipelock blocked request: {data.get('block_reason')}")
    return data.get("content", "")

TLS Interception#

To enable it:

Generate a CA and enable TLS interception (see the TLS Interception Guide)
Trust the CA in your Python environment:

export SSL_CERT_FILE=~/.pipelock/ca.pem
# Or for requests/httpx specifically:
export REQUESTS_CA_BUNDLE=~/.pipelock/ca.pem

MCP proxy mode (stdio wrapping) does not require TLS interception. It scans
traffic in both directions without certificates.

Choosing a Config#

Preset	Action	Best For
`generic-agent`	warn	New agent integrations (recommended starting point)
`balanced`	warn	General purpose, fetch proxy tuning
`claude-code`	block	Claude Code, unattended agents
`strict`	block	High-security, production

Start with generic-agent to log detections without blocking. Review the
logs, tune thresholds, then switch to strict for production.

Troubleshooting#

MCP server not starting#

Verify the command works without Pipelock first:

npx -y @modelcontextprotocol/server-filesystem /tmp

Then wrap it:

pipelock mcp proxy -- npx -y @modelcontextprotocol/server-filesystem /tmp

Seeing Pipelock output#

Pipelock logs to stderr. In CrewAI, stderr from stdio subprocesses may be
captured. To see real-time output during development:

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | \
  pipelock mcp proxy --config pipelock.yaml -- npx -y @modelcontextprotocol/server-filesystem /tmp

False positives#

Switch to warn mode to see what's being flagged without blocking:

response_scanning:
  action: warn
mcp_input_scanning:
  action: warn
mcp_tool_scanning:
  action: warn

Review stderr output, then tighten thresholds.

Config file not found#

Use absolute paths if relative paths don't resolve in your CrewAI project
structure:

MCPServerStdio(
    command="pipelock",
    args=["mcp", "proxy", "--config", "/etc/pipelock/config.yaml", "--",
          "npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
)

Using Pipelock with CrewAI#

Quick Start#

How It Works#

Integration Patterns#

Pattern A: mcps= DSL (CrewAI >= 1.4.0)#

Pattern B: MCPServerAdapter (crewai-tools)#

Pattern C: Multiple servers with mixed transports#

What Pipelock Catches#

Docker Compose#

HTTP Fetch Proxy#

TLS Interception#

Choosing a Config#

Troubleshooting#

MCP server not starting#

Seeing Pipelock output#

False positives#

Config file not found#

Using Pipelock with CrewAI#

Quick Start#

How It Works#

Integration Patterns#

Pattern A: mcps= DSL (CrewAI >= 1.4.0)#

Pattern B: MCPServerAdapter (crewai-tools)#

Pattern C: Multiple servers with mixed transports#

What Pipelock Catches#

Docker Compose#

HTTP Fetch Proxy#

TLS Interception#

Choosing a Config#

Troubleshooting#

MCP server not starting#

Seeing Pipelock output#

False positives#

Config file not found#

Pattern A: `mcps=` DSL (CrewAI >= 1.4.0)#

Pattern B: `MCPServerAdapter` (crewai-tools)#

Pattern A: `mcps=` DSL (CrewAI >= 1.4.0)#

Pattern B: `MCPServerAdapter` (crewai-tools)#