CrewAI

Gate every CrewAI tool call through LatchGate’s enforcement pipeline.

Prerequisites

• LatchGate running (latchgate up or production deployment)

Install

pip install latchgate-crewai

Quick start

from crewai import Agent, Task, Crew
from latchgate_crewai import LatchGateToolset

lg = LatchGateToolset.create_sync(gate_url="http://localhost:3000")
tools = lg.all()

agent = Agent(
    role="Secure Worker",
    goal="Perform tasks through gated tools with full audit trail",
    backstory="You are an agent with access to LatchGate-protected actions.",
    tools=tools,
)

task = Task(
    description="Fetch https://httpbin.org/get and report the response",
    expected_output="The HTTP response body",
    agent=agent,
)

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

How it works

LatchGateToolset discovers all registered LatchGate actions via the REST API and wraps each one as a CrewAI BaseTool. When the LLM calls a tool, the wrapper routes it through LatchGateClient.execute() — the full enforcement pipeline runs transparently.

Use create_sync() from synchronous code or create() from async code. Both accept the same parameters.

Filtering actions

lg = LatchGateToolset.create_sync(
    gate_url="http://localhost:3000",
    include={"http_fetch", "github_read"},
    exclude={"send_message"},
)

Async usage

async with await LatchGateToolset.create(gate_url="http://localhost:3000") as lg:
    tools = lg.all()
    # use tools...

UDS transport (production)

from latchgate import LatchGateClient

client = LatchGateClient(socket="/run/latchgate/gate.sock", agent_id="my-crewai-agent")
lg = LatchGateToolset.create_sync(client=client)

Audit callback

Use on_audit for programmatic access to receipt metadata after each successful execution:

def my_audit_handler(record):
    # record.action_id, record.receipt_id, record.trace_id, record.verification
    audit_log.store(record)

lg = LatchGateToolset.create_sync(
    gate_url="http://localhost:3000",
    on_audit=my_audit_handler,
)

LATCHGATE_URL environment variable

You can set LATCHGATE_URL instead of passing gate_url explicitly:

# With LATCHGATE_URL=http://localhost:3000 set:
lg = LatchGateToolset.create_sync()

Discovery timeout

Control the HTTP timeout for the discovery phase (default 15 seconds):

lg = LatchGateToolset.create_sync(
    gate_url="http://localhost:3000",
    discovery_timeout=10.0,
)

More details

Full API reference: latchgate-integrations/crewai/README.md.