Add Cryptographic Identity to Your LangChain Agents

Prerequisites: Python 3.10+, langchain installed

The problem

Your LangChain executors call tools anonymously. When a supervisor routes a task to a specialist agent, 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 executor 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 LangChain adapter for AIP:

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

Create a plugin, build your executor as normal, then register it with a name. The name becomes the agent's AIP identity key:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from aip_agents import AIPConfig
from aip_agents.adapters.langchain import AIPLangChainPlugin

# Create the plugin
plugin = AIPLangChainPlugin(AIPConfig(app_name="my-langchain-app"))

# Build your executor as normal
@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

llm = ChatOpenAI(model="gpt-4o")
agent = create_openai_tools_agent(llm, [search], prompt)
executor = AgentExecutor(agent=agent, tools=[search])

# Register AIP -- this is the only line you add
plugin.register(executor, name="researcher")

# 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(executor, name="researcher"), AIP:

• Generated an Ed25519 keypair for the researcher identity
• Assigned it an AIP identifier (aip:key:ed25519:z6Mkf5...)
• Extracted the tool names from the executor and used them as the token scope
• Issued a signed compact token and stored it for retrieval

The token flow looks like this:

Agent (researcher)
  --> AIPLangChainPlugin.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 the assigned scope for any registered agent:

scope = plugin.get_agent_scope("researcher")
print(f"Scope: {scope}")
# Scope: ['search']  -- derived from the executor's tool list

For supervisor patterns, register all agents at once with register_agents(), then create delegation tokens between them. The delegation token narrows scope from supervisor to specialist:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from aip_agents import AIPConfig
from aip_agents.adapters.langchain import AIPLangChainPlugin

plugin = AIPLangChainPlugin(AIPConfig(
    app_name="my-langchain-app",
    log_tokens=True,  # prints identity and delegation events
))

# Build individual executors
@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

@tool
def write(content: str) -> str:
    """Write a document."""
    return f"Written: {content}"

llm = ChatOpenAI(model="gpt-4o")

researcher_executor = AgentExecutor(
    agent=create_openai_tools_agent(llm, [search], researcher_prompt),
    tools=[search],
)
writer_executor = AgentExecutor(
    agent=create_openai_tools_agent(llm, [write], writer_prompt),
    tools=[write],
)

# Register all agents at once (supervisor pattern)
plugin.register_agents({
    "researcher": researcher_executor,
    "writer": writer_executor,
})

# researcher delegates to writer with attenuated scope
delegation_token = plugin.create_delegation(
    parent_name="researcher",
    child_name="writer",
    task_description="Write a summary of the research findings",
    scope=["write"],  # narrower than researcher's full scope
)

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

Each call to create_delegation() records the parent, the child, and the attenuated scope. The child's token is updated in place, so subsequent calls to get_tool_call_headers("writer") return the delegated token rather than the original.

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"]
# 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.langchain import AIPLangChainPlugin

# Issue a token that expired 10 seconds ago
plugin = AIPLangChainPlugin(AIPConfig(
    app_name="my-langchain-app",
    token_ttl_seconds=-10,  # negative TTL = already expired
))
plugin.register(executor, name="researcher")

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