Add Cryptographic Identity to Your CrewAI Agents

Prerequisites: Python 3.10+, crewai installed

The problem

Your CrewAI agents call tools anonymously. When the researcher delegates to the writer, no identity verification happens. The tool server has no way to confirm which agent is making the request, let alone whether that agent was authorized to do so. If a tool server is compromised, any agent can impersonate any other. There is no audit trail of who authorized what.

AIP gives each agent a self-certifying Ed25519 keypair and an AIP identifier. Every tool call carries a signed token that records the issuing agent, the permitted scope, the budget ceiling, and the delegation depth. The tool server verifies the signature before serving the request. If the token is missing, expired, or out of scope, the request is rejected.

Install the CrewAI adapter for AIP:

This pulls in agent-identity-protocol (core SDK) and the CrewAI integration layer. For other frameworks: aip-agents[adk], aip-agents[langchain], or aip-agents[all].

Create a plugin, build your crew as normal, then register the plugin. One call wires identity into every agent:

from crewai import Agent, Crew, Task, Process
from aip_agents import AIPConfig
from aip_agents.adapters.crewai import AIPCrewPlugin

# Create the plugin
plugin = AIPCrewPlugin(AIPConfig(app_name="my-research-app"))

# Define agents as you normally would
researcher = Agent(
    role="researcher",
    goal="Find accurate information about AI agent identity protocols",
    backstory="Expert at web research and source verification",
    tools=[],
    allow_delegation=False,
)

writer = Agent(
    role="writer",
    goal="Write clear, concise summaries of research findings",
    backstory="Technical writer specializing in AI systems",
    tools=[],
    allow_delegation=False,
)

# Build the crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[
        Task(
            description="Research the current state of agent identity protocols",
            expected_output="A summary of existing protocols and gaps",
            agent=researcher,
        ),
        Task(
            description="Write a blog post based on the research",
            expected_output="A 500-word blog post",
            agent=writer,
        ),
    ],
    process=Process.sequential,
)

# Register AIP -- this is the only line you add
plugin.register(crew)

# Get auth headers for outgoing tool calls
headers = plugin.get_tool_call_headers("researcher")
print(headers)
# {"X-AIP-Token": "eyJhbGciOiJFZERTQSIsInR5cCI6IkpXVCJ9..."}

When you called plugin.register(crew), AIP walked every agent in the crew and:

• Generated an Ed25519 keypair for each agent
• Assigned each agent an AIP identifier (aip:key:ed25519:z6Mkf5...)
• Issued a signed compact token scoped to that agent's tools
• Stored the token so get_tool_call_headers() can retrieve it on demand

The token flow looks like this:

Agent (researcher)
  --> AIPCrewPlugin.get_tool_call_headers("researcher")
      --> returns {"X-AIP-Token": "<signed-jwt>"}
          --> outgoing request to MCP server
              --> server verifies signature + scope
              --> serves request or returns 401

You can inspect any agent's identity directly:

identity = plugin.identity_manager.get("researcher")
print(f"AIP ID: {identity.aip_id}")
# AIP ID: aip:key:ed25519:z6Mkf5rGM...

When the researcher hands off a task to the writer, that handoff can be cryptographically recorded. The delegation token carries a narrowed scope -- the writer can only exercise the permissions explicitly granted, not the researcher's full token:

from crewai import Agent, Crew, Task, Process
from aip_agents import AIPConfig
from aip_agents.adapters.crewai import AIPCrewPlugin

plugin = AIPCrewPlugin(AIPConfig(
    app_name="my-research-app",
    log_tokens=True,  # prints delegation events to stdout
))

researcher = Agent(role="researcher", goal="...", backstory="...", tools=[])
writer = Agent(role="writer", goal="...", backstory="...", tools=[])
editor = Agent(role="editor", goal="...", backstory="...", tools=[])

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[...],
    process=Process.sequential,
)
plugin.register(crew)

# researcher delegates to writer with attenuated scope
delegation_token = plugin.create_delegation(
    parent_role="researcher",
    child_role="writer",
    task_description="Write summary based on research findings",
    scope=["write", "summarize"],  # narrower than researcher's full scope
)

# writer further delegates to editor with even narrower scope
plugin.create_delegation(
    parent_role="writer",
    child_role="editor",
    task_description="Proofread the draft",
    scope=["edit"],
)

depth = plugin.token_manager.chain_depth(delegation_token)
print(f"Delegation chain depth: {depth}")
# Delegation chain depth: 1

Each delegation step records the parent, the child, and the attenuated scope. The chain depth increments at each hop. You can set max_depth in AIPConfig to enforce a ceiling -- any attempt to delegate beyond that depth is rejected by the token verifier.

from aip_token.compact import CompactToken

token = plugin.get_tool_call_headers("researcher")["X-AIP-Token"]
tampered = token[:-5] + "XXXXX"

# On the server side:
# CompactToken.verify(tampered, public_key_bytes) raises:
#   SignatureError: signature verification failed

# researcher token is scoped to ["search", "fetch"]
# Trying to call a tool outside that scope:
#
# Server checks: "tool:email" not in token.scope
# Server returns: 401 Unauthorized -- scope mismatch

from aip_agents import AIPConfig
from aip_agents.adapters.crewai import AIPCrewPlugin

# Issue a token that expired 10 seconds ago
plugin = AIPCrewPlugin(AIPConfig(
    app_name="my-research-app",
    token_ttl_seconds=-10,  # negative TTL = already expired
))
plugin.register(crew)

token = plugin.get_tool_call_headers("researcher")["X-AIP-Token"]
# CompactToken.verify raises: TokenExpiredError