External Agents (Pro-Code)

Connect LangGraph, CrewAI, PydanticAI, AutoGen, and custom agents

Builder Guide: External Agents

This guide covers best practices for building external agents that are compatible with the platform. Whether you're using LangGraph, CrewAI, AutoGen, Pydantic AI, n8n, or a custom framework, this guide shows you how to structure your endpoints so the platform can forward messages, apply governance, and track cost and performance.

Key concept: The platform is the governance layer (guardrails, access control, HITL, audit, tracing). Your external agent is the execution layer (LLM inference, tool calling, reasoning). Studio creates a wrapper that connects the two.

Table of Contents

• How It Works
• AG-UI + A2A Protocol (Turn-Based)
• Supported Input Formats
• Building a LangServe-Compatible Agent
• Building an OpenAI-Compatible Agent
• Building an n8n Webhook Agent
• Building a Custom Format Agent
• Reporting Token Usage for FinOps
• Health Check Endpoints
• Authentication
• MCP Tools with External Agents
  • Tool Discovery (AG-UI + A2A)
• MCP Tools with n8n Workflows
• Observability and Tracing
• Streaming Intermediate Events
• Error Handling
• Testing Your Integration
• Platform LLM Proxy
• A2A Self-Registration
• Demo: LangGraph Agent
• Demo: Research Pipeline (Full Integration)
• Demo: Agent Examples

How It Works

When a user chats with an external agent through the Portal:

User (Portal)
  |
  v
API Gateway (L1)
  |-- Input guardrails (PII detection, prompt injection)
  |-- 4D access control checks
  |
  v
External Agent Executor
  |-- Build request body (format-specific)
  |-- Add auth headers
  |-- HTTP POST to your endpoint
  |-- Normalize response
  |
  v
Your External Agent
  |-- Receives messages
  |-- Runs LLM inference, tool calls, reasoning
  |-- Returns response
  |
  v
API Gateway (post-processing)
  |-- Output guardrails
  |-- Cost estimation (if costModel + token usage reported)
  |-- Persist message, record execution, record trace
  |-- Audit event
  |-- Stream response to user via SSE

The platform applies input guardrails before forwarding to your agent, and output guardrails after receiving the response. Your agent never needs to implement guardrails -- the platform handles it.

AG-UI + A2A Protocol (Turn-Based)

For agents that need per-tool observability, HITL approval gates, and fine-grained retry, the platform offers the AG-UI + A2A dual-protocol model — a turn-based HTTP+JSON protocol.

The difference: In legacy mode, your entire agent pipeline runs as a single opaque Temporal activity. With the dual-protocol model, each tool call becomes a separate Temporal activity, giving you:

• Per-tool retry with exponential backoff
• Per-tool HITL approval gates (workflow pauses until human approves)
• Per-tool visibility in the Temporal UI
• Workflow restart from the last successful activity (not from scratch)
• Full audit trail of every tool execution

How to adopt

1. Add a POST /turn endpoint to your agent that accepts AgentTurnRequest and returns AgentTurnResponse
2. Set turnBasedEnabled: true in your agent's externalConfig
3. Return toolCalls[] declarations instead of executing tools directly — the platform executes them

Quick example

@app.post("/turn")
async def turn(body: TurnRequest):
    if body.turnNumber == 0:
        return {
            "messages": [{"role": "assistant", "content": "Searching..."}],
            "toolCalls": [{"id": "tc-1", "name": "web_search", "arguments": {"query": "..."}}],
            "nextAction": "await_tool_results",
        }
    # Process tool results
    results = body.toolResults or []
    return {
        "messages": [{"role": "assistant", "content": f"Found: {results}"}],
        "nextAction": "done",
    }

Framework examples

The examples/agent-examples/ directory contains complete implementations for 4 frameworks:

All agents are pre-seeded on the platform and run as Docker containers via docker compose up.

Backward compatibility

The dual-protocol model is fully backward compatible. If turnBasedEnabled is false (default), your agent runs in legacy mode with no changes needed. You can have both /invoke and /turn on the same agent.

For the full protocol specification, types, state management, and HITL details, see the AG-UI + A2A protocol documentation.

Supported Input Formats

The platform can format requests in five ways, selected in the Identity tab:

Building a LangServe-Compatible Agent

This is the most common format for Python-based agents using LangGraph.

Request your agent receives

{
  "input": {
    "messages": [
      { "role": "system", "content": "You are a helpful assistant..." },
      { "role": "user", "content": "What is 2+2?" },
      { "role": "assistant", "content": "4" },
      { "role": "user", "content": "And 3+3?" }
    ]
  },
  "config": {},
  "kwargs": {}
}

Response your agent should return

{
  "output": "The answer is 6.",
  "metadata": {
    "usage": {
      "prompt_tokens": 150,
      "completion_tokens": 25
    }
  }
}

The platform extracts:

• Message: from output (string) or output.messages[-1].content
• Token usage: from metadata.usage, callback_events[].usage, or __run.usage

Minimal Python Example (FastAPI)

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Message(BaseModel):
    role: str
    content: str

class InvokeInput(BaseModel):
    messages: list[Message] = []

class InvokeRequest(BaseModel):
    input: InvokeInput
    config: dict = {}
    kwargs: dict = {}

@app.post("/invoke")
async def invoke(request: InvokeRequest):
    messages = request.input.messages
    last_message = messages[-1].content if messages else ""

    # Your agent logic here (LangGraph, tool calls, etc.)
    response = f"I processed: {last_message}"

    return {
        "output": response,
        "metadata": {
            "usage": {
                "prompt_tokens": 100,  # Report actual token counts
                "completion_tokens": 20,
            }
        }
    }

@app.get("/health")
async def health():
    return {"status": "healthy"}

Best Practices for LangServe Agents

1. Always include a health check endpoint (GET /health) so the Deploy tab can show connection status.
2. Report token usage in metadata so the platform can calculate cost. Without this, FinOps dashboards show $0.
3. Handle the system message. The platform prepends system context (from the Prompt tab) as the first message with role: "system". Your agent can use or ignore it.
4. Return a string in output. The platform expects either a plain string or an object with a messages array where it takes the last message's content.
5. Keep responses under the timeout. Default is 30 seconds. For long-running agents, increase the timeout in the Identity tab.

Building an OpenAI-Compatible Agent

If your agent exposes an OpenAI-compatible /chat/completions endpoint:

Request your agent receives

{
  "model": "your-model-name",
  "messages": [
    { "role": "system", "content": "..." },
    { "role": "user", "content": "Hello" }
  ],
  "temperature": 0.5
}

Response your agent should return

{
  "choices": [
    {
      "message": {
        "role": "assistant",
        "content": "Hello! How can I help you?"
      }
    }
  ],
  "usage": {
    "prompt_tokens": 50,
    "completion_tokens": 12
  }
}

This format works with any OpenAI-compatible inference server (vLLM, Ollama, text-generation-inference, etc.).

Building an n8n Webhook Agent

For n8n workflows triggered via webhook:

Request your workflow receives

{
  "message": "Check the status of ticket #1234",
  "conversationHistory": [
    { "role": "user", "content": "Previous message" },
    { "role": "assistant", "content": "Previous response" }
  ],
  "agentId": "agent-5",
  "userId": "user-1",
  "conversationId": "conv-abc"
}

Response your workflow should return

{
  "response": "Ticket #1234 is currently open and assigned to the DevOps team.",
  "tokenUsage": {
    "input": 200,
    "output": 30
  }
}

The platform tries multiple field names for the response: response, output, message, text, result, content, answer, reply.

Best Practices for n8n Workflows

1. Use the Webhook trigger node as your workflow entry point. Set it to accept POST requests.
2. Return a JSON response from the last node. The platform looks for the message in several common fields (see above).
3. Include token usage if your workflow calls an AI node. n8n's AI nodes often expose token counts. Pass them through to the response as tokenUsage: { input, output }.
4. Keep the webhook URL stable. If you change it, update the agent's External Config in Studio.
5. The platform falls back to LLM if your n8n workflow is unreachable (HTTP error, timeout). The Fallback Prompt (Prompt tab) is used in this case.

Connecting MCP Tools to n8n Workflows

n8n workflows can invoke the platform's MCP tools through the HTTP Request node:

POST http://api-gateway:4000/api/tools/{toolId}/invoke
Headers:
  Authorization: Bearer <agent-token>
  X-Agent-Id: <agent-id>
Body:
  { "input": { "query": "search term" } }

This allows n8n workflows to use knowledge base search, ticket creation, SAST scanning, and other platform-registered MCP tools while respecting tool grants and access control.

Building a Custom Format Agent

For agents that don't fit the standard formats:

Request your agent receives

{
  "message": "User's latest message",
  "conversationHistory": [
    { "role": "user", "content": "..." },
    { "role": "assistant", "content": "..." }
  ]
}

Response Extraction

Set a Response Extractor (JSONPath) in the Identity tab to tell the platform where to find the message in your response. For example:

• If your response is { "data": { "reply": "Hello" } }, set the extractor to data.reply.
• If your response is { "result": { "text": "Hello" } }, set the extractor to result.text.

Without a response extractor, the platform tries these fields in order: message, output, response, text, result, content, answer, reply, then falls back to stringifying the entire response.

Reporting Token Usage for FinOps

The platform cannot count tokens for external agents because it only sees the final response, not the internal LLM calls, tool chains, retries, or RAG context your agent assembles. Token reporting must come from the external agent.

How cost estimation works

1. Your agent reports token counts in the response (format-specific fields).
2. You configure a Cost Model in the Identity tab (e.g., claude-sonnet-4, gpt-4o).
3. The platform multiplies your reported token counts by the configured model's pricing.

If your agent doesn't report tokens

Cost will show as $0 in FinOps dashboards. This is accurate (the platform genuinely doesn't know), not a bug.

Token usage field locations by format

Best practice: aggregate across internal steps

If your agent makes multiple LLM calls (e.g., planning + execution + summarization), sum the tokens across all calls and report the total:

total_input = plan_tokens.input + exec_tokens.input + summary_tokens.input
total_output = plan_tokens.output + exec_tokens.output + summary_tokens.output
return {
    "output": final_response,
    "metadata": {
        "usage": {
            "prompt_tokens": total_input,
            "completion_tokens": total_output,
        }
    }
}

Health Check Endpoints

The Deploy tab polls your health check URL to show connection status. Implement a simple endpoint:

@app.get("/health")
async def health():
    return {"status": "healthy"}

The platform considers any 2xx response as healthy. If the endpoint returns a non-2xx status or times out (5 seconds), the Deploy tab shows a red status indicator.

If you don't specify a separate Health Check URL, the platform uses your main endpoint URL.

Authentication

Three auth modes are supported:

The authSecretRef is a reference to a credential stored in the platform's tenant_credentials table or as an environment variable. It is never the raw secret itself.

MCP Tools with External Agents

External pro-code agents can call the platform's MCP tools by making HTTP requests back to the API gateway. This keeps tool access governed by the platform's access control layer.

From a LangGraph agent (Python)

import httpx

async def call_platform_tool(tool_id: str, input_data: dict, agent_id: str, token: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"http://api-gateway:4000/api/tools/{tool_id}/invoke",
            json={"input": input_data},
            headers={
                "Authorization": f"Bearer {token}",
                "X-Agent-Id": agent_id,
            },
        )
        return response.json()

# Example: Search the knowledge base
result = await call_platform_tool(
    "tool-kb",
    {"query": "password reset policy"},
    agent_id="agent-5",
    token="<a2a-invocation-secret>",
)

The MCP gateway enforces tool grants: if the agent hasn't been granted access to tool-kb in the Tools tab, the call returns 403.

Tool Discovery (AG-UI + A2A)

When using the AG-UI + A2A protocols (turn-based mode), your agent automatically receives its available tools on turn 0 via the availableTools field. This eliminates the need to hardcode tool names and schemas.

@app.post("/turn")
async def turn(body: TurnRequest):
    if body.turnNumber == 0 and body.availableTools:
        # Platform sent the tools this agent can use
        for tool in body.availableTools:
            print(f"Tool: {tool.name} — {tool.description}")
            if tool.conditions and tool.conditions.get("requireApproval"):
                print(f"  ⚠ Requires HITL approval")

        # Use tool schemas to configure your LLM framework dynamically
        tool_schemas = [
            {"name": t.name, "description": t.description, "parameters": t.inputSchema}
            for t in body.availableTools
        ]

The availableTools array includes only tools the agent has been granted access to (D2 enforcement). Each tool includes its name, description, JSON input schema, permissions, and conditions (rate limits, HITL requirements). See the AG-UI + A2A protocol docs for the full specification.

MCP Tools with n8n Workflows

n8n workflows can use the platform's MCP tools through the HTTP Request node. This is useful when you want an n8n workflow to search the knowledge base, create tickets, or use any other platform-registered tool.

Step-by-step setup

1. Add an HTTP Request node to your n8n workflow.
2. Configure the request:
   • Method: POST
   • URL: http://api-gateway:4000/api/tools/{toolId}/invoke
   • Headers: Authorization: Bearer <token>, X-Agent-Id: <agent-id>
   • Body (JSON): { "input": { "query": "search term" } }
3. Grant the tool to the agent in Studio's Tools tab. The MCP gateway enforces default-deny.

Example: Knowledge base search from n8n

In your n8n workflow, add an HTTP Request node with:

• URL: http://api-gateway:4000/api/tools/tool-kb/invoke
• Method: POST
• Body: { "input": { "query": "{{ $json.message }}" } }
• Headers: Authorization: Bearer {{ $env.AGENT_TOKEN }}

The response contains the search results, which you can process in subsequent n8n nodes.

Example: Creating a ticket from n8n

• URL: http://api-gateway:4000/api/tools/tool-ticket/invoke
• Method: POST
• Body: { "input": { "title": "New ticket", "priority": "high", "description": "..." } }

All tool invocations are logged, rate-limited, and subject to HITL if configured on the tool grant.

Observability and Tracing

The platform provides full observability for external agents without any instrumentation on your side:

What the platform traces automatically

What shows up in the Trace Explorer

Every external agent execution appears in Studio > Traces with:

• Multi-span waterfall view (context retrieval -> external call -> guardrails)
• Token usage (if reported by your agent)
• Cost (if costModel is configured and tokens are reported)
• Error categorization and messages
• Duration breakdown per span

OTel export

If you run a Jaeger instance (included in docker-compose), external agent spans are exported via OpenTelemetry alongside platform-managed agent spans.

Streaming Intermediate Events

By default, external agents are treated as a black box: the platform sends one HTTP POST and waits for the complete JSON response. For long-running agents (research pipelines, multi-step reasoning, tool-heavy workflows), this means the user sees nothing until the final answer arrives — which can take 60-120 seconds.

Streaming mode lets your external agent emit intermediate events (pipeline progress, tool calls, partial text) that the platform relays to the Portal frontend in real-time. The user sees step-by-step progress, collapsible tool cards, and streamed report text — identical to platform-managed agents.

How it works

User (Portal)
  |
  v
API Gateway
  |-- Sends Accept: text/event-stream header
  |
  v
Your Agent /invoke
  |-- Detects Accept header
  |-- Returns StreamingResponse (text/event-stream)
  |-- Yields SSE events as pipeline progresses
  |
  v
API Gateway (streaming relay)
  |-- Relays token, workflow_step, tool_call events to frontend
  |-- Collects spans for traces
  |-- On "done" event: runs output guardrails, calculates cost, persists
  |
  v
Portal (real-time rendering)
  |-- Step-by-step progress bar
  |-- Collapsible tool cards
  |-- Streamed report text

No new endpoints, no callbacks, no webhooks. Same /invoke URL, same request body, same auth — your agent just returns a streaming response instead of JSON when the platform asks for it.

Enabling streaming

Two changes are needed:

1. Platform side (Studio): Enable streaming in the agent's External Config:

   {
     "endpointUrl": "http://your-agent:8200/invoke",
     "inputFormat": "langserve",
     "streamingEnabled": true,
     "streamingTimeoutMs": 300000
   }
   

   • streamingEnabled (default: false): When true, the platform sends Accept: text/event-stream
   • streamingTimeoutMs (default: 300000): Maximum time to wait for events (5 minutes)

2. Agent side: Check the Accept header and return SSE when requested:

   from fastapi import FastAPI, Request
   from starlette.responses import StreamingResponse
   
   @app.post("/invoke")
   async def invoke(request: Request, body: InvokeRequest):
       accept = request.headers.get("accept", "")
       if "text/event-stream" in accept:
           return StreamingResponse(
               stream_pipeline(body),
               media_type="text/event-stream",
               headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
           )
       # Fallback: return JSON as before
       return {"output": "...", "metadata": {...}}
   

SSE Event Protocol (9 types)

Your agent emits JSON events, one per data: line. Each event has a type field:

1. workflow_step — Pipeline progress

Emitted when a pipeline node starts or completes. The frontend renders a step-by-step progress bar.

data: {"type": "workflow_step", "step": {"index": 0, "name": "Research Briefer", "type": "llm_call", "status": "running"}}

data: {"type": "workflow_step", "step": {"index": 0, "name": "Research Briefer", "type": "llm_call", "status": "completed", "durationMs": 4200, "output": {"questions_generated": 4}}}

Fields:

• step.index — Step number (0-based)
• step.name — Human-readable step name
• step.type — "llm_call", "tool_execution", or "processing"
• step.status — "running", "completed", "failed", or "waiting_approval"
• step.durationMs — Time taken (on completion)
• step.output — Optional summary of step output

2. tool_call — Tool invocation

Emitted when your agent calls a tool. The frontend renders a collapsible tool card.

data: {"type": "tool_call", "toolCall": {"id": "s1", "name": "web_search", "status": "success", "input": {"query": "quantum computing breakthroughs 2025"}, "output": {"title": "Nature: Quantum...", "url": "https://..."}, "durationMs": 800}}

Fields:

• toolCall.id — Unique tool call ID
• toolCall.name — Tool name (e.g., "web_search", "code_exec")
• toolCall.status — "success" or "error"
• toolCall.input — Tool input parameters
• toolCall.output — Tool output (truncated if large)
• toolCall.durationMs — Time taken

3. token — Streamed output text

Emitted as the final text is generated. The frontend accumulates tokens into the message.

data: {"type": "token", "content": "The research "}
data: {"type": "token", "content": "findings indicate "}
data: {"type": "token", "content": "that quantum computing "}

Emit tokens at word or sentence boundaries for natural rendering. Do not emit character-by-character.

4. done — Final event (REQUIRED)

Every streaming response must end with a done event. The platform uses this for persistence, cost tracking, and output guardrails.

data: {"type": "done", "message": "# Research Report\n\nFull report text here...", "tokenUsage": {"input": 8729, "output": 4531, "cost": 0}, "metadata": {"agent": "research-pipeline", "sources_found": 12, "duration_seconds": 87.3}}

Fields:

• message — The complete final text (used for persistence and output guardrails)
• tokenUsage.input — Total input tokens across all LLM calls
• tokenUsage.output — Total output tokens across all LLM calls
• tokenUsage.cost — Pre-calculated cost (0 if using platform cost model)
• metadata — Any additional metadata for the trace

5. error — Terminal error

Emitted when the agent encounters an unrecoverable error. The platform stops and reports to the frontend.

data: {"type": "error", "error": "LLM API timeout after 30s in summarizer node"}

6. guardrail — Guardrail notification

Emitted if your agent runs its own guardrails (optional — the platform runs guardrails regardless).

data: {"type": "guardrail", "guardrail": {"name": "pii_check", "passed": true, "action": "none"}}

7. thinking_complete — LLM thinking/reasoning (platform-emitted)

Emitted by the platform when the LLM narrates its plan before calling tools. The frontend renders a collapsible "Thinking" block so users can distinguish reasoning from the final answer.

data: {"type": "thinking_complete", "content": "I'll analyze the CSV using pandas. Let me first read the file...", "iteration": 1}

Fields:

• content — The LLM's reasoning text (displayed in a dimmed, collapsible block)
• iteration — Which LLM iteration produced this thinking (for multi-step workflows)

8. tool_status — Tool execution sub-status (platform-emitted)

Emitted during tool execution to show granular progress. Currently used for code_execute to indicate sandbox lifecycle stages. The frontend updates the tool card badge.

data: {"type": "tool_status", "toolCallId": "tc-123", "subStatus": "Starting sandbox..."}

Fields:

• toolCallId — The tool call this status belongs to
• subStatus — Human-readable status message (e.g., "Starting sandbox...", "Executing code...")

9. approval_required — HITL approval gate (platform-emitted)

Emitted when a tool invocation requires human approval before execution. The frontend renders an approval card with approve/reject buttons.

data: {"type": "approval_required", "approval": {"id": "approval-abc123", "action": "Execute tool: create_ticket", "toolName": "create_ticket", "toolInput": {"title": "..."}, "riskScore": 70, "impact": "data", "reasoning": "Tool requires human approval before execution."}}

After the user approves via POST /api/approvals/:id/decide, the platform re-executes the tool and emits tool_result and workflow_step events to update the UI.

Python helper function

A minimal helper for emitting SSE events from any Python framework:

import json

def sse_event(event_type: str, **kwargs) -> str:
    """Format a single SSE event."""
    return f"data: {json.dumps({'type': event_type, **kwargs})}\n\n"

# Usage in a generator:
def stream_pipeline(topic: str):
    yield sse_event("workflow_step", step={"index": 0, "name": "Planning", "status": "running"})
    result = do_planning(topic)
    yield sse_event("workflow_step", step={"index": 0, "name": "Planning", "status": "completed"})

    for word in result.split(" "):
        yield sse_event("token", content=word + " ")

    yield sse_event("done", message=result, tokenUsage={"input": 100, "output": 50, "cost": 0})

LangGraph example

For LangGraph agents, run each node individually instead of graph.invoke() to emit events between nodes:

def stream_langgraph_pipeline(topic: str):
    state = {"topic": topic, ...}

    # Run each node and emit events between them
    nodes = [
        ("Planner", planner_node),
        ("Researcher", researcher_node),
        ("Writer", writer_node),
    ]

    for idx, (name, node_fn) in enumerate(nodes):
        yield sse_event("workflow_step", step={
            "index": idx, "name": name, "type": "llm_call", "status": "running"
        })
        result = node_fn(state)
        state.update(result)
        yield sse_event("workflow_step", step={
            "index": idx, "name": name, "type": "llm_call", "status": "completed"
        })

    report = state["final_report"]
    for word in report.split(" "):
        yield sse_event("token", content=word + " ")

    yield sse_event("done", message=report, tokenUsage={...})

CrewAI example

Use CrewAI's step_callback to emit events as each task completes:

from crewai import Crew

def stream_crewai(topic: str):
    step_idx = 0

    def on_step(step_output):
        nonlocal step_idx
        yield sse_event("workflow_step", step={
            "index": step_idx,
            "name": step_output.task.description[:60],
            "status": "completed",
        })
        step_idx += 1

    crew = Crew(agents=[...], tasks=[...], step_callback=on_step)
    result = crew.kickoff(inputs={"topic": topic})

    yield sse_event("done", message=str(result), tokenUsage={...})

Note: CrewAI's step_callback is synchronous. For true streaming, wrap the crew execution in a thread and use a queue to pass events to the generator.

Testing streaming with curl

# Test SSE streaming directly against your agent:
curl -N -X POST http://localhost:8200/invoke \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "input": {"messages": [{"role": "user", "content": "Research quantum computing"}]},
    "config": {},
    "kwargs": {}
  }'

# You should see events streaming in real-time:
# data: {"type": "workflow_step", "step": {"index": 0, "name": "Research Briefer", "status": "running"}}
# data: {"type": "workflow_step", "step": {"index": 0, "name": "Research Briefer", "status": "completed", ...}}
# data: {"type": "tool_call", "toolCall": {"id": "s1", "name": "web_search", ...}}
# data: {"type": "token", "content": "The research "}
# data: {"type": "token", "content": "findings indicate "}
# ...
# data: {"type": "done", "message": "...", "tokenUsage": {...}}

Backward compatibility

Streaming is fully backward-compatible:

• streamingEnabled: false (default): Platform sends normal POST, agent returns JSON. No change.
• streamingEnabled: true, agent returns JSON: The platform detects Content-Type: application/json and falls back to the non-streaming path automatically.
• streamingEnabled: true, agent returns SSE: Full streaming with real-time events.

This means you can enable streaming in the platform config before your agent supports it — the platform gracefully handles JSON responses from agents that haven't been updated yet.

What the platform handles

When streaming is active, the platform still provides full governance:

Error Handling

The platform categorizes errors from external agents:

Best practices

1. Return meaningful error messages. The platform passes them through to the user.
2. Use appropriate HTTP status codes. 500 for internal errors, 503 for temporary unavailability, 408 for your own timeouts.
3. Don't swallow errors silently. If your agent encounters an error, return it explicitly rather than returning a generic response.
4. Set a reasonable timeout. Complex LangGraph agents with multiple tool calls may need 60-120 seconds. Simple chatbots should work within 10-30 seconds.

Testing Your Integration

1. Start your external agent

# LangGraph example
cd examples/langgraph-agent
pip install -r requirements.txt
python app.py
# Agent available at http://localhost:8100

2. Create an agent in Studio

1. Go to Studio > Agents > Create Agent.
2. In the Identity tab, select External Pro-Code.
3. Set Endpoint URL to http://localhost:8100/invoke.
4. Set Input Format to LangServe / LangGraph.
5. Save the agent.

3. Test via Portal

1. Go to Portal > Chat.
2. Select your agent.
3. Send a message. You should see the response from your external agent.

4. Verify observability

1. Go to Studio > Traces.
2. Find the trace for your conversation.
3. Verify the span hierarchy shows: context retrieval -> external call -> guardrails.

5. Test via curl

curl -X POST http://localhost:4000/api/chat \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer mock-token" \
  -d '{
    "agentId": "<your-agent-id>",
    "message": "Hello from the API"
  }'

Platform LLM Proxy

The platform provides an OpenAI-compatible LLM proxy that external agents can route their LLM calls through. This is the most impactful integration point: it puts every intermediate LLM call under platform governance, not just the final chat boundary.

The problem it solves

A complex agent (like a research pipeline) may make 10+ internal LLM calls during a single task: planning, summarizing, compressing, report writing. Without the proxy, these intermediate calls:

• Bypass input/output guardrails (PII could leak, injections aren't caught)
• Are invisible to model access policies (agent can use any model)
• Don't appear in cost tracking (FinOps shows $0 for intermediate work)
• Can't be audited or traced

How to use it

Route your LLM calls through POST /api/llm/chat/completions instead of calling AIFS/OpenAI directly. The proxy accepts the standard OpenAI request format:

# Instead of:
url = "https://api.openai.com/v1/chat/completions"
headers = {"Authorization": f"Bearer {OPENAI_KEY}"}

# Use:
url = "http://api-gateway:4000/api/llm/chat/completions"
headers = {"Authorization": f"Bearer {INVOCATION_SECRET}"}  # a2a_inv_* token

The proxy applies this pipeline on every call:

1. Agent identity -- resolved from a2a_inv_* token or X-Agent-Id header
2. Model access check -- verifies the agent is allowed to use the requested model
3. Input guardrails -- PII detection, prompt injection, jailbreak scanning
4. Forward to AIFS -- calls the actual LLM provider
5. Output guardrails -- toxicity, hallucination, off-topic detection (non-streaming)
6. Cost attribution -- calculates cost and emits per-agent metrics

Authentication

The proxy requires agent identity. Three options:

Endpoints

Streaming support

The proxy supports "stream": true. For streaming requests, input guardrails run before forwarding but output guardrails are skipped (since the response arrives in chunks).

A2A Self-Registration

External agents can programmatically register themselves on the platform using the Google A2A JSON-RPC 2.0 protocol. This eliminates manual agent creation in Studio -- the agent announces itself on startup.

Note: The legacy REST endpoints (/api/a2a/register, /api/a2a/invoke, /api/a2a/discover) have been removed. Use the Google A2A JSON-RPC protocol at POST /api/a2a/v1 for all agent-to-agent communication.

How it works

1. The agent serves a spec-compliant Agent Card at GET /.well-known/agent.json
2. Registration is performed via POST /api/a2a/v1 with JSON-RPC method agent/register
3. The platform creates the agent with status draft (pending admin approval)
4. The platform returns an invocation secret (a2a_inv_*) -- shown once
5. After admin approval (status -> active), the agent can authenticate with the secret

Registration request

curl -X POST http://localhost:4000/api/a2a/register \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My Research Agent",
    "description": "LangGraph research pipeline",
    "version": "1.0.0",
    "capabilities": ["research", "web-search"],
    "callbackUrl": "http://localhost:8200/invoke",
    "callbackAuth": {"type": "none"},
    "tenantId": "tenant-1",
    "registrationToken": "<token-from-admin>"
  }'

Registration response

{
  "data": {
    "agentId": "agent-abc12345",
    "invocationSecret": "a2a_inv_7f3a...",
    "status": "pending_approval",
    "a2aInvokeUrl": "http://localhost:4000/api/a2a/invoke"
  }
}

Using the invocation secret

The invocationSecret is the agent's permanent credential. Use it as a Bearer token:

# LLM proxy
curl -X POST http://localhost:4000/api/llm/chat/completions \
  -H "Authorization: Bearer a2a_inv_7f3a..." \
  -H "Content-Type: application/json" \
  -d '{"model":"claude-sonnet-4","messages":[{"role":"user","content":"hello"}]}'

# Tool invoke
curl -X POST http://localhost:4000/api/tools/tool-web-search/invoke \
  -H "Authorization: Bearer a2a_inv_7f3a..." \
  -d '{"input":{"query":"quantum computing"}}'

# A2A invoke (call another agent)
curl -X POST http://localhost:4000/api/a2a/invoke \
  -H "Authorization: Bearer a2a_inv_7f3a..." \
  -d '{"targetAgentId":"agent-xyz","payload":{"message":"summarize this"}}'

The platform resolves the agent's identity from the invocation secret automatically -- no X-Agent-Id header needed.

Python example (on startup)

import httpx

async def self_register():
    resp = await httpx.AsyncClient().post(
        "http://api-gateway:4000/api/a2a/register",
        json={
            "name": "My Agent",
            "capabilities": ["research"],
            "callbackUrl": "http://localhost:8200/invoke",
            "callbackAuth": {"type": "none"},
            "tenantId": "tenant-1",
            "registrationToken": os.environ["A2A_REGISTRATION_TOKEN"],
        },
    )
    data = resp.json()["data"]
    print(f"Registered as {data['agentId']}")
    print(f"Secret: {data['invocationSecret']}")  # Save this!

Demo: LangGraph Agent

A complete demo agent is included in examples/langgraph-agent/:

• app.py -- FastAPI server with LangServe-compatible /invoke endpoint, calculator tool, and health check
• requirements.txt -- Python dependencies (fastapi, uvicorn, pydantic)
• Dockerfile -- Container build for docker-compose

Running the demo

cd examples/langgraph-agent
pip install -r requirements.txt
python app.py

The agent starts at http://localhost:8100 with:

• POST /invoke -- LangServe-format chat endpoint
• POST /chat -- Generic/custom format endpoint
• GET /health -- Health check

Configure it in Studio with:

• Execution Mode: External Pro-Code
• Endpoint URL: http://localhost:8100/invoke
• Input Format: LangServe / LangGraph

Demo: Research Pipeline (Full Integration)

The examples/research-agent/ directory contains a fully integrated LangGraph agent that demonstrates all three platform integration features: LLM Proxy, MCP Tool Invocation, and A2A Self-Registration.

Architecture

┌─────────────────────────────────────────────────────┐
│  Research Pipeline (LangGraph StateGraph)            │
│                                                      │
│  briefer → lead_researcher → researcher → summarizer │
│                  ↑                |        → compressor │
│                  └── loop ───────┘        → reporter    │
│                                                      │
│  LLM calls ──→ Platform LLM Proxy ──→ AIFS          │
│                  (guardrails + cost tracking)         │
│                                                      │
│  Web search ──→ Platform Tool Invoke ──→ MCP Gateway │
│                  (D2 grant enforcement)               │
│                                                      │
│  Startup ──→ A2A Self-Registration ──→ Agent record  │
│                  (invocation secret)                  │
└─────────────────────────────────────────────────────┘

Three operating modes

Quick start (full platform mode)

cd examples/research-agent
pip install -r requirements.txt

# 1. Get an invocation secret (or registration token)
export PLATFORM_AGENT_TOKEN=a2a_inv_...

# 2. Point LLM and tool calls at the platform
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

# 3. Optional: self-register on startup
export A2A_REGISTRATION_TOKEN=<token-from-admin>
export PLATFORM_BASE_URL=http://localhost:4000

# 4. Run
python app.py

What happens at each stage

1. Startup: If A2A_REGISTRATION_TOKEN is set, the agent registers via Google A2A JSON-RPC and prints its invocation secret.

2. LLM calls (llm_chat()): Every call to Claude/GPT goes through POST /api/llm/chat/completions. The platform applies:

   • Input guardrails (PII detection, prompt injection)
   • Model access check (D4 — is this agent allowed this model?)
   • Cost attribution (tokens tracked per agent)
   • Output guardrails (non-streaming)

3. Web search (web_search()): If PLATFORM_TOOL_URL is configured, calls POST /api/tools/{toolId}/invoke. The platform applies:

   • D2 tool grant check (is this agent granted tool-web-search?)
   • HITL approval if the tool requires it (handles 202 Accepted)
   • Falls back to local DuckDuckGo if the platform returns 403 (tool not granted)

4. Response metadata: The /invoke response includes platform_tracking with mode and features enabled.

Key code patterns

Auth headers — every platform call includes the invocation secret:

def _platform_headers():
    return {
        "Authorization": f"Bearer {_PLATFORM_AGENT_TOKEN}",
        "X-Agent-Id": _AGENT_ID,
        "Content-Type": "application/json",
    }

Graceful fallback — if a tool invoke returns 403, fall back to local:

resp = httpx.post(url, headers=_platform_headers(), json=payload, timeout=30)
if resp.status_code == 403:
    return _local_web_search(query, max_results)  # D2 not granted, use local

Self-registration — fire-and-forget on startup:

@app.on_event("startup")
async def on_startup():
    if os.environ.get("A2A_REGISTRATION_TOKEN"):
        await self_register()

SSE streaming

The research agent supports real-time streaming. When the platform sends Accept: text/event-stream, the agent streams intermediate events:

• workflow_step events for each pipeline node (Research Briefer, Lead Researcher, Web Search, Summarizer, Compressor, Final Reporter)
• tool_call events for each web search result found
• token events for the final report text word-by-word
• done event with the complete report, token usage, and metadata

To enable: set streamingEnabled: true in the agent's External Config (already enabled in seed data).

See Streaming Intermediate Events for the full protocol specification.

Endpoints

Demo: Agent Examples

The examples/agent-examples/ directory contains four fully functioning agents implementing the AG-UI + A2A protocols. Each uses different platform MCP tools and demonstrates a different framework pattern.

Running with Docker Compose

All four agents are included in docker-compose.yml and start automatically:

docker compose up -d
# Agents available at ports 8300-8303

Testing the HITL flow (CrewAI agent)

The CrewAI agent demonstrates a complete HITL approval flow:

1. Send a message to the CrewAI agent via the Portal or API
2. The agent searches KB and web, then requests create_ticket
3. The platform pauses the Temporal workflow (ticket requires HITL approval)
4. Approve or reject via POST /api/approvals/:id/decide
5. The workflow resumes: tool executes (or rejection is sent to agent), agent composes final response

The approval decision signals the Temporal workflow directly — the workflow resumes within seconds of approval.

Verifying in Temporal UI

Open http://localhost:8080 to see the workflow history. Each agent's workflow shows individual activities for every turn and tool call, not a single opaque blob.

Related Documentation

• AG-UI + A2A Protocol -- Turn-based protocol specification, types, state management, and HITL details
• Registering Agents Guide -- Full agent designer walkthrough (all 9 tabs)
• Tool Integration Guide -- MCP tool registration and granting
• Builder Access Control Guide -- Configuring all 4 access control dimensions
• Architecture Reference -- Platform layer details
• LLM Provider Configuration -- T-Systems AIFS models and pricing