Framework Integration

LangGraph, CrewAI, PydanticAI, AutoGen, n8n guidance

Framework Integration Guide

This guide covers how to integrate specific agent frameworks -- LangGraph, CrewAI, PydanticAI, AutoGen, n8n, and raw Python -- with the Enterprise Agentic AI Platform. Each section includes architecture notes, integration patterns, real code examples from the examples/ directory, and best practices.

Prerequisite: Read AGENT_CONTRACT.md for the full turn-based protocol spec and EXTERNAL_AGENTS.md for connection patterns and supported input formats.

Table of Contents

• AG-UI + A2A Protocol Overview
• LangGraph Integration
• CrewAI Integration
• PydanticAI Integration
• AutoGen Integration
• n8n Integration
• Raw Python (No Framework)
• Common Patterns Across All Frameworks

AG-UI + A2A Protocol Overview

The AG-UI + A2A dual-protocol model is a turn-based protocol where your agent declares tool calls and the platform executes them. This split gives the platform per-tool observability, HITL approval gates, retry, and audit -- regardless of which framework your agent uses.

Protocol summary:

1. Platform sends POST /turn with { threadId, messages, context, turnNumber, availableTools }.
2. Your agent returns { messages, toolCalls, nextAction, statePatch, tokenUsage }.
3. If nextAction is "await_tool_results", the platform executes each tool as a separate Temporal activity and calls /turn again with the results.
4. The loop continues until nextAction is "done" or maxTurns is reached.

Every tool execution becomes a distinct Temporal activity with D2 grant enforcement, HITL gates, retry with backoff, and full audit trail. Without the dual-protocol model, the entire agent pipeline runs as a single opaque activity.

Tool discovery: On turn 0, the platform sends availableTools -- the list of tools this agent has been granted access to, including names, JSON input schemas, and conditions (e.g., requireApproval: true for HITL tools). Agents can use this to dynamically build tool schemas instead of hardcoding them.

LangGraph Integration

Architecture: LangGraph models agent logic as a StateGraph -- a directed graph of nodes connected by edges and conditional routing. Each node is a Python function that receives and returns typed state. This is well suited for multi-step research pipelines, iterative reasoning, and complex branching logic.

Legacy SSE Mode vs AG-UI + A2A Mode

The platform supports two integration modes for LangGraph agents:

The minimal LangGraph agent in examples/langgraph-agent/app.py demonstrates legacy mode with a LangServe-compatible /invoke endpoint. The research pipeline in examples/research-agent/app.py demonstrates full platform integration with LLM proxy routing, MCP tool invocation, and A2A self-registration.

Example: Research Pipeline

The research pipeline (examples/research-agent/app.py) implements a 6-node StateGraph:

research_briefer -> lead_researcher -> researcher (web_search) -> summarizer -> compressor -> final_reporter

State definition -- typed with TypedDict:

class ResearchState(TypedDict):
    topic: str
    research_questions: list[str]
    current_question: str
    current_question_index: int
    raw_sources: list[dict[str, Any]]
    summaries: list[str]
    compressed_notes: str
    iterations: int
    max_iterations: int
    research_complete: bool
    final_report: str
    errors: list[str]

Graph construction:

graph = StateGraph(ResearchState)

graph.add_node("research_briefer", research_briefer)
graph.add_node("lead_researcher", lead_researcher)
graph.add_node("researcher", researcher)
graph.add_node("summarizer", summarizer)
graph.add_node("compressor", compressor)
graph.add_node("final_reporter", final_reporter)

graph.set_entry_point("research_briefer")
graph.add_edge("research_briefer", "lead_researcher")
graph.add_edge("lead_researcher", "researcher")
graph.add_edge("researcher", "summarizer")
graph.add_edge("summarizer", "compressor")

graph.add_conditional_edges(
    "compressor",
    should_continue,
    {
        "lead_researcher": "lead_researcher",
        "final_reporter": "final_reporter",
    },
)
graph.add_edge("final_reporter", END)

Three operating modes controlled by environment variables:

# 1. Standalone -- direct LLM + local DuckDuckGo
export LLM_API_KEY=your-api-key
python app.py

# 2. Platform Proxy -- LLM calls via platform proxy (guardrails + cost tracking)
export PLATFORM_AGENT_TOKEN=a2a_inv_...
export PLATFORM_LLM_URL=http://localhost:4000/api/llm
python app.py

# 3. Full Platform -- proxy + MCP tools + A2A self-registration
export PLATFORM_AGENT_TOKEN=a2a_inv_...
export PLATFORM_LLM_URL=http://localhost:4000/api/llm
export PLATFORM_TOOL_URL=http://localhost:4000/api/tools
export PLATFORM_TOOL_WEB_SEARCH_ID=tool-web-search
python app.py

Best Practices

• Use TypedDict for graph state to get IDE autocompletion and type checking across nodes.
• Keep individual graph nodes focused -- one responsibility per node (search, summarize, compress, report).
• Use conditional_edges with a router function for control flow rather than complex branching logic inside nodes.
• Route LLM calls through the Platform LLM Proxy (PLATFORM_LLM_URL) so every intermediate call gets guardrails, model access checks, and cost attribution.
• For AG-UI + A2A mode, serialize LangGraph state into statePatch so the platform's Temporal workflow can persist it between turns.

CrewAI Integration

Architecture: CrewAI uses role-based multi-agent crews where each Agent has a role, goal, and backstory. Agents work on Task objects, coordinated by a Crew using a process strategy (sequential or hierarchical).

Mapping CrewAI Concepts to the Platform

Key pattern: CrewAI has no native "declare but don't execute" mode. The integration uses a reasoning-only crew pattern -- CrewAI agents run with tools=[] (no tools registered on the crew). The crew's structured JSON output is parsed to construct toolCalls[] for the platform.

Example: Support Ticket Crew

The CrewAI example (examples/agent-examples/crewai_agent.py) implements a support ticket workflow with three agent personas:

• Researcher -- triages issues, determines search strategy
• Analyst -- synthesizes findings, decides on escalation
• TicketCreator -- composes detailed support tickets

Turn flow:

Turn 0: Researcher crew -> analyze issue -> declare knowledge_base_search
Turn 1: Process KB results -> declare web_search
Turn 2: Analyst crew -> synthesize -> declare create_ticket (HITL!)
Turn 3: Process ticket result (approved/rejected) -> done

CrewAI LLM setup -- uses crewai.LLM to route through the Platform LLM Proxy:

from crewai import LLM as CrewLLM

_crewai_llm = CrewLLM(
    model=LLM_MODEL,
    base_url=_effective_base_url,
    api_key=_effective_api_key,
)

Reasoning crew execution on turn 0 (Researcher persona):

from crewai import Agent as CrewAgent, Crew, Process, Task as CrewTask

researcher = CrewAgent(
    role="Support Researcher",
    goal="Analyze customer support issues and determine the best investigation strategy",
    backstory="You are a seasoned support researcher...",
    llm=_crewai_llm,
    verbose=False,
    allow_delegation=False,
)

task = CrewTask(
    description=f"Analyze this customer support issue...\nRespond with ONLY valid JSON...",
    expected_output="JSON with category, severity, search_query, needs_ticket",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[task], process=Process.sequential, verbose=False)
result = crew.kickoff()
analysis = _parse_json(result.raw)

HITL Flow Demonstration

The create_ticket tool has requireApproval: true in its grant conditions. When the Analyst crew decides to escalate, the agent declares a create_ticket tool call. The platform pauses the Temporal workflow, sends an approval request to the Portal UI, and waits for a human reviewer. The agent handles both outcomes:

if ctx.get("step") == "ticket_create":
    for tr in tool_results:
        if tr.get("status") == "success":
            ticket_id = output.get("ticketId", "unknown")
            ticket_result = f"Ticket created: **#{ticket_id}**"
        elif tr.get("status") == "rejected":
            reason = tr.get("rejectionReason", "No reason given")
            ticket_result = f"Ticket creation was rejected by reviewer: {reason}"

Best Practices

• Set allow_delegation=False on agents to prevent CrewAI from spawning uncontrolled delegation chains.
• Set verbose=False to reduce log noise in production.
• Have the crew output structured JSON (enforced via task description) so you can reliably parse it into toolCalls[].
• Use separate crews for separate reasoning phases (triage vs. synthesis vs. ticket writing) rather than one large crew.
• Dependencies: crewai>=0.86.0 (see requirements-crewai.txt). Use Dockerfile.crewai to isolate the dependency tree.

PydanticAI Integration

Architecture: PydanticAI provides structured, type-safe agent interactions with first-class support for "declare but don't execute" tool calling via ExternalToolset and DeferredToolRequests.

ExternalToolset and DeferredToolRequests Pattern

PydanticAI has the cleanest fit with the AG-UI + A2A protocols. Its ExternalToolset is a first-class "declare but don't execute" primitive:

1. Define tools as ToolDefinition objects (name + JSON schema). No tool implementation code.
2. When the LLM decides to call a tool, Agent.run() returns DeferredToolRequests with call IDs, names, and arguments.
3. On the next turn, feed results back via DeferredToolResults + message_history.

Example: Knowledge Search Agent

The PydanticAI example (examples/agent-examples/pydantic_ai_agent.py) implements a knowledge search agent that queries an internal KB and supplements with web search.

ExternalToolset setup:

from pydantic_ai import Agent as PydanticAgent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai.tools import ToolDefinition
from pydantic_ai.toolsets import ExternalToolset

_provider = OpenAIProvider(base_url=_effective_base_url, api_key=_effective_api_key)
_model = OpenAIChatModel(LLM_MODEL, provider=_provider)

_external_toolset = ExternalToolset([
    ToolDefinition(
        name="knowledge_base_search",
        description="Search the internal knowledge base for documents matching a query.",
        parameters_json_schema={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"},
                "limit": {"type": "integer", "description": "Max results", "default": 5},
            },
            "required": ["query"],
        },
    ),
    ToolDefinition(
        name="web_search",
        description="Search the web using DuckDuckGo for current information.",
        parameters_json_schema={...},
    ),
])

_pydantic_agent = PydanticAgent(
    _model,
    system_prompt="You are a knowledgeable research assistant...",
    toolsets=[_external_toolset],
)

Turn 0 -- initial run:

result = await agent.run(query, output_type=[str, DeferredToolRequests])

if isinstance(result.output, DeferredToolRequests):
    tool_calls = []
    for call in result.output.calls:
        tool_calls.append({
            "id": call.tool_call_id,
            "name": call.tool_name,
            "arguments": call.args_as_dict(),
        })
    # Serialize message history for state persistence
    msgs_json = result.all_messages_json()
    return {
        "toolCalls": tool_calls,
        "nextAction": "await_tool_results",
        "statePatch": {"pydantic_messages": msgs_json.decode()},
    }

Turn 1+ -- feeding results back:

from pydantic_ai import DeferredToolResults
from pydantic import TypeAdapter
from pydantic_ai.messages import ModelMessage

ta = TypeAdapter(list[ModelMessage])
prev_messages = ta.validate_json(prev_messages_json)

deferred = DeferredToolResults()
for tr in tool_results:
    call_id = tr.get("toolCallId", "")
    if tr.get("status") == "success":
        deferred.calls[call_id] = json.dumps(output) if not isinstance(output, str) else output
    elif tr.get("status") == "rejected":
        deferred.calls[call_id] = ModelRetry(f"Tool rejected: {tr.get('rejectionReason')}")

result = await _pydantic_agent.run(
    message_history=prev_messages,
    output_type=[str, DeferredToolRequests],
    deferred_tool_results=deferred,
)

Dynamic tool discovery -- when the platform sends availableTools on turn 0, the agent builds a new ExternalToolset dynamically:

def _build_dynamic_agent(available_tools):
    tool_defs = [
        ToolDefinition(
            name=t.name,
            description=t.description,
            parameters_json_schema=t.inputSchema or {"type": "object", "properties": {}},
        )
        for t in available_tools
    ]
    toolset = ExternalToolset(tool_defs)
    return PydanticAgent(_model, system_prompt="...", toolsets=[toolset])

Best Practices

• Use ExternalToolset exclusively -- never register tool implementations locally. The platform handles all execution.
• Serialize message history via result.all_messages_json() into statePatch for Temporal state persistence.
• Deserialize with TypeAdapter(list[ModelMessage]).validate_json() on subsequent turns.
• Handle DeferredToolRequests vs plain str output using output_type=[str, DeferredToolRequests].
• Use ModelRetry for rejected tools so PydanticAI can attempt an alternative strategy.
• Dependencies: pydantic-ai>=0.0.36 (see requirements-pydantic-ai.txt). Use Dockerfile.pydantic-ai for isolation.

AutoGen Integration

Architecture: AutoGen uses conversation-based multi-agent patterns. The platform integration uses OpenAIChatCompletionClient for LLM-driven tool selection (getting FunctionCall objects back) and AssistantAgent for final reasoning synthesis.

OpenAIChatCompletionClient Setup

AutoGen's model client lets you call the LLM with tool schemas and receive FunctionCall objects without executing them -- a natural fit for the AG-UI + A2A "declare, don't execute" model.

from autogen_ext.models.openai import OpenAIChatCompletionClient

_model_client = OpenAIChatCompletionClient(
    model=LLM_MODEL,
    base_url=_effective_base_url,
    api_key=_effective_api_key,
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "family": "unknown",
        "structured_output": True,
    },
)

Example: Code Review Agent

The AutoGen example (examples/agent-examples/autogen_agent.py) implements a code review agent with SAST scanning, JSON validation, and database schema lookup.

Tool schemas are plain dictionaries (no implementations needed):

_SAST_SCHEMA = {
    "name": "sast_scanner",
    "description": "Run a static application security testing (SAST) scan on a code snippet...",
    "parameters": {
        "type": "object",
        "properties": {
            "code": {"type": "string", "description": "Source code to scan"},
            "language": {"type": "string", "description": "Programming language", "default": "auto"},
        },
        "required": ["code"],
    },
}

Turn 0 -- LLM call with tool schemas:

from autogen_core import CancellationToken, FunctionCall
from autogen_core.models import SystemMessage, UserMessage, CreateResult

messages = [
    SystemMessage(content=SYSTEM_PROMPT),
    UserMessage(content=prompt, source="user"),
]

result: CreateResult = await _model_client.create(
    messages=messages,
    tools=active_tool_schemas,
    cancellation_token=CancellationToken(),
)

if isinstance(result.content, list) and result.content:
    tool_calls = []
    for fc in result.content:
        if isinstance(fc, FunctionCall):
            args = json.loads(fc.arguments)
            tool_calls.append({"id": fc.id, "name": fc.name, "arguments": args})
    return {"toolCalls": tool_calls, "nextAction": "await_tool_results", ...}

Turn 1+ -- feeding results back via FunctionExecutionResultMessage:

from autogen_core.models import (
    AssistantMessage as AutoGenAssistantMessage,
    FunctionExecutionResult,
    FunctionExecutionResultMessage,
)

# Reconstruct assistant's tool call message
function_calls = [FunctionCall(id=fc["id"], name=fc["name"], arguments=fc["arguments"])
                   for fc in prev_fc_json]
messages.append(AutoGenAssistantMessage(content=function_calls, source="assistant"))

# Add tool results
exec_results = [
    FunctionExecutionResult(call_id=call_id, content=content, name=tool_name, is_error=False)
    for ...
]
messages.append(FunctionExecutionResultMessage(content=exec_results))

# Call the model again
result = await _model_client.create(messages=messages, tools=active_tool_schemas, ...)

Final synthesis using AssistantAgent -- pure reasoning, no tools:

from autogen_agentchat.agents import AssistantAgent

synthesis_agent = AssistantAgent(
    name="SeniorReviewer",
    model_client=_model_client,
    tools=[],  # No tools -- pure reasoning
    system_message="You are a senior software engineer performing code reviews...",
)
result = await synthesis_agent.run(task=task)

Best Practices

• Use OpenAIChatCompletionClient.create() with tool schemas rather than AutoGen's full agent loop, so the platform controls tool execution.
• Serialize both the conversation messages and the FunctionCall objects into statePatch for state persistence.
• Use model_info to declare capabilities accurately -- set function_calling: True for tool use.
• Leverage AssistantAgent (with tools=[]) for final synthesis steps that need pure LLM reasoning.
• Dependencies: autogen-agentchat>=0.4.0 and autogen-ext[openai]>=0.4.0 (see requirements-autogen.txt). Use Dockerfile.autogen for isolation.

n8n Integration

n8n workflows integrate with the platform through a webhook-based pattern and a set of four sidecar services that provide governance, memory, observability, and identity.

Webhook-Based Workflow Design

Every n8n workflow that acts as a platform agent must follow this pattern:

[Webhook Trigger] --> [Your Logic] --> [Respond to Webhook]

The Webhook Trigger accepts { message, agentId, userId, conversationId } from the platform. The Respond to Webhook node returns JSON with a response field.

Using Templates

The platform provides five built-in templates (template guide coming soon):

The ai-chat-agent and mcp-connected-agent templates are the most common for building conversational agents. Both pre-configure the LLM Chat Model node to route through the Guardrails Proxy.

Sidecar Configuration

The n8n Bridge Layer consists of four sidecars that run alongside the n8n container:

LLM Chat Model Node Routing

The critical configuration is routing all LLM calls through the Guardrails Proxy. In docker-compose.yml, n8n is configured with:

N8N_AI_OPENAI_API_BASE: http://guardrails-proxy:8081/v1

In the LLM Chat Model node within n8n, set:

• Base URL: http://guardrails-proxy:8081/v1
• Model: claude-sonnet-4 (or any model available via T-Systems AIFS)
• Temperature: 0.3 (recommended default)

This ensures every LLM call from an n8n workflow passes through the platform's guardrails.

Best Practices

• Always use the Guardrails Proxy for LLM calls -- never configure direct API keys in n8n nodes.
• Use the mcp-connected-agent template when your workflow needs to call platform-registered MCP tools.
• Use the basic-webhook template as a starting point for custom logic that does not need LLM.
• Ensure your Respond to Webhook node returns a field named response, message, output, text, or result -- the platform checks these in order.
• For MCP tool access, connect the MCP Client Tool node to the Identity Injector MCP Proxy at http://identity-injector:8085.

Raw Python (No Framework)

When you need full control with minimal dependencies, a raw Python agent implements the AG-UI + A2A protocols directly with just FastAPI and httpx.

Minimal Protocol Implementation

The raw Python agent (examples/agent-examples/raw_python_agent.py) is the simplest possible AG-UI + A2A protocol implementation. It uses no framework -- just FastAPI for the HTTP server, httpx for LLM calls, and Pydantic for request validation.

Request model:

class AgentTurnRequest(BaseModel):
    threadId: str
    messages: list[AgentMessage] = []
    toolResults: list[dict[str, Any]] | None = None
    context: dict[str, Any] = {}
    turnNumber: int = 0
    availableTools: list[AvailableToolInfo] | None = None

Turn flow:

Turn 0: Parse question -> declare web_search tool call
Turn 1: Receive search results -> declare text_summarizer tool call
Turn 2: Receive summary -> compose final answer with LLM -> done

Turn 0 -- declaring a tool call:

@app.post("/turn")
async def turn(request: Request, body: AgentTurnRequest):
    if body.turnNumber == 0:
        return {
            "messages": [{"role": "assistant", "content": f"Searching the web for: **{query}**"}],
            "toolCalls": [{
                "id": f"ws-{int(time.time())}",
                "name": "web_search",
                "arguments": {"query": query, "limit": 5},
            }],
            "nextAction": "await_tool_results",
            "statePatch": {"query": query, "step": "searching"},
            "tokenUsage": {"input": 0, "output": 0, "cost": 0},
        }

Health endpoint (required for all agents):

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "agent": "raw-python-research-summarize",
        "version": "1.0.0",
        "contract": "turn-based",
        "tools_used": ["web_search", "text_summarizer"],
        "llm_configured": bool(LLM_API_KEY),
    }

Dockerfile -- minimal, no framework dependencies:

FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir fastapi uvicorn httpx
COPY agent-examples/ .
ENV PORT=8300
CMD ["sh", "-c", "python ${AGENT_FILE:-raw_python_agent.py}"]

When to Use Raw Python vs a Framework

Common Patterns Across All Frameworks

Platform LLM Proxy for Intermediate Calls

Every framework agent can route LLM calls through the Platform LLM Proxy to get guardrails, model access checks, and cost attribution on every intermediate call -- not just the final chat response.

# When PLATFORM_AGENT_TOKEN and PLATFORM_LLM_URL are set:
PLATFORM_AGENT_TOKEN = os.environ.get("PLATFORM_AGENT_TOKEN", "")
PLATFORM_LLM_URL = os.environ.get("PLATFORM_LLM_URL", "")

if PLATFORM_AGENT_TOKEN and PLATFORM_LLM_URL:
    LLM_API_KEY = PLATFORM_AGENT_TOKEN
    LLM_BASE_URL = PLATFORM_LLM_URL

The proxy endpoint is OpenAI-compatible (/chat/completions), so all frameworks (CrewAI, PydanticAI, AutoGen, LangGraph) can use it as a drop-in base_url replacement.

Tool Discovery via availableTools

On turn 0, the platform sends availableTools with the tools this agent has been granted. All example agents check for this field and build tool schemas dynamically:

if body.turnNumber == 0 and body.availableTools:
    tool_names = [t.name for t in body.availableTools]
    print(f"Available tools from platform: {tool_names}")
    # Build framework-specific tool schemas from body.availableTools

This allows agents to adapt to their granted tool set rather than hardcoding tool definitions. Agents fall back to hardcoded defaults when availableTools is absent (backward compatibility).

A2A Self-Registration

External agents can self-register on the platform at startup using the Google A2A JSON-RPC protocol. This creates a platform agent record with a persistent identity.

Note: The legacy REST endpoint (/api/a2a/register) has been removed. Use POST /api/a2a/v1 with JSON-RPC 2.0. SDK agents handle registration automatically.

async def self_register() -> str | None:
    url = f"{_PLATFORM_BASE_URL}/api/a2a/v1"
    body = {
        "name": "Research Pipeline Agent",
        "description": "Multi-agent LangGraph research pipeline...",
        "version": "2.0.0",
        "capabilities": ["research", "web-search", "summarization"],
        "callbackUrl": f"http://localhost:{port}/invoke",
        "callbackAuth": {"type": "none"},
        "tenantId": tenant_id,
        "registrationToken": reg_token,
    }
    resp = await client.post(url, json=body, timeout=10.0)
    data = resp.json().get("data", {})
    agent_id = data.get("agentId", "")
    invocation_secret = data.get("invocationSecret", "")
    return agent_id

The registration token is pre-issued by an admin. The returned invocation secret can be used as PLATFORM_AGENT_TOKEN for future runs.

Cost Model Configuration

All agents should report token usage in their turn responses for FinOps attribution:

return {
    "messages": [...],
    "nextAction": "done",
    "tokenUsage": {
        "input": _total_input_tokens,
        "output": _total_output_tokens,
        "cost": 0,  # Platform calculates cost from model pricing
    },
}

Framework-specific extraction patterns:

• PydanticAI: result.usage() returns request_tokens and response_tokens.
• CrewAI: result.token_usage provides prompt_tokens and completion_tokens.
• AutoGen: result.usage has prompt_tokens and completion_tokens.
• Raw Python: Track tokens manually from the OpenAI-compatible response usage field.

Health Endpoint Requirements

Every agent must expose a GET /health endpoint. The platform uses this for deployment health checks, liveness probes, and the Studio Deploy tab status indicator.

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "agent": "your-agent-name",
        "version": "1.0.0",
        "contract": "turn-based",
        "tools_used": ["tool_a", "tool_b"],
    }

The response must include "status": "healthy" for the platform to consider the agent available. Including contract, tools_used, and version is recommended but not required.

Dockerfile Isolation

Each framework has its own Dockerfile to keep dependency trees isolated:

All agents default to mock mode when no LLM_API_KEY or PLATFORM_AGENT_TOKEN is set, returning hardcoded responses that demonstrate the turn flow without requiring a real LLM.