Agent DocsDeveloper documentation
Reference

SDK Guide (Python & TypeScript)

Python and TypeScript SDK reference with examples

Agentic Platform SDK Developer Guide

Comprehensive guide for building, testing, and deploying AI agents on the Agentic Platform.

Table of Contents


Overview

The Agentic Platform SDK lets you build external agents in Python or TypeScript that integrate with the platform's governance layer. Your agent handles the reasoning; the platform handles tool execution, guardrails, memory, tracing, and access control.

Your Code (Agent Logic)
    │
    ▼
SDK (AgenticAgent base class)
    │  • HTTP routing (FastAPI / Hono)
    │  • Serialization (types, helpers)
    │  • Agent card (/.well-known/agent.json)
    │  • Health checks
    │
    ▼
Platform (Orchestrator)
    │  • Tool execution via MCP
    │  • Input/output guardrails
    │  • Memory (short-term, long-term, graph)
    │  • Trace spans (OpenTelemetry)
    │  • Access control (Cedar policies)
    │  • HITL approval workflows
    │
    ▼
End Users (Agent Portal)

The Dual-Protocol Model (AG-UI + A2A)

Agents communicate with the platform using two standard protocols:

  • AG-UI (Server-Sent Events) — Real-time streaming of text, tool calls, and state between agent and platform
  • A2A (JSON-RPC 2.0) — Agent discovery via Agent Cards at /.well-known/agent.json
Scroll to zoom · Drag to pan · 100%

AG-UI Message Flow

The platform sends POST /ag-ui with conversation messages and available tools. Your agent streams SSE events back:

  • TEXT_MESSAGE_CONTENT — Text deltas for the response
  • TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END — Declare tool calls
  • STATE_SNAPSHOT / STATE_DELTA — Share agent state
  • RUN_FINISHED — Signal turn completion

The SDK handles AG-UI serialization internally. Your agent implements on_turn() — the SDK's bridge layer converts AG-UI messages to/from the turn format automatically.

Why "Agent Declares, Platform Executes"?

This model provides:

  • Security: The agent never has direct tool access; the platform enforces grants and rate limits
  • Governance: Every tool call is logged, traced, and auditable
  • Guardrails: The platform can intercept and filter tool inputs/outputs
  • HITL: High-risk tool calls can require human approval before execution

State Ownership Contract

The platform and the agent own different slices of state. Violating this split silently causes accumulated-history bugs (state leaking from turn to turn, duplicated messages, runaway response sizes on A2A delegation). The contract is:

OwnerWhat
Platformtask.messages — conversational history; tool_results; hitl_response; workflow steps; cost/token accounting. Carried forward turn-to-turn on the platform side and handed to the agent as input on every call.
Agent — per turnExactly one of: (a) a new assistant reply, (b) a HITL/surface request, (c) one or more declared tool calls. Must NOT include the cumulative conversation — just what the agent produced THIS turn.
Agent — opaque state_patchAgent-specific scratchpad the platform echoes back on the next turn (e.g. research_questions, current_question_index, phase). Opaque to the platform; the agent owns its shape. Must NOT be used for messages, tool_calls, next_action, state_patch (self), tool_results, turn_number, or _token_usage — these are protocol fields excluded from auto-capture.
Agent — privateInternal graph state (LangGraph checkpointer, AutoGen thread, CrewAI memory, etc.). Implementation detail. Must NOT leak into the turn response.

Under this contract, a call to on_turn() is a single-turn function: given inputs = (prior messages + tool results + hitl response), produce output = (one new reply / tool calls / HITL request) + optional state_patch. The agent never appends to the platform's conversational history directly.

LangGraph subtlety. When the adapter runs graph.ainvoke(input_state, config={thread_id: ...}), LangGraph's checkpointer rehydrates prior state including state.messages. The adapter automatically snapshots the incoming message count and returns only messages added during THIS invocation — the delta. Your graph nodes can append to state.messages freely for internal reasoning; only the per-turn delta reaches the platform. Use a dedicated output field (e.g. final_report) and LangGraphAdapter(graph, output_key="final_report") if you want an explicit single-message reply.

A2A delegation. The same contract holds. A sub-agent invoked via invoke_agent runs one logical call per orchestrator turn; the sub-agent's reply appears in the orchestrator's tool result. History is carried forward at the orchestrator's level, not the sub-agent's — the sub-agent is re-invoked fresh for each delegation turn, with any agent-specific state it needs supplied via state_patch.

HITL Response Contract

When your agent pauses with next_action="await_approval" and emits an A2UI surface (via surface_request(...), picker_surface(...), etc.), the user's submission comes back in context.hitl_response on the next on_turn call. The contract is the v0.9 action envelope, verbatim:

hitl_response = {
  "action": {
    "kind": "event",
    "name": "urn:a2ui:submit",   # the URN your surface's submit button used
    "context": {...}              # the submit-context dict YOUR surface defined
  },
  "dataModel": {...} | None,     # optional full-state snapshot if your
                                  # createSurface had sendDataModel=true
  "respondedBy": "<user-id>",
  "respondedAt": "<iso-ts>",
}

The agent reads hitl_response.action.context.<field> — keyed on the shape the agent itself put on the submit button when it built the surface. No legacy type tags (type: "hitl_plan_selection_response"), no parallel field names (selectedIds vs selected). The platform passes the envelope through unchanged; your agent's code is the authority on how to interpret its own surface's submit.

Convention (not enforcement) for the four standard surface helpers:

Surface helperSubmit context shapeAgent reads
picker_surface{requestId, selected: [ids…]}ctx["selected"]
input_surface{requestId, value: "…"}ctx["value"]
form_surface{requestId, fields: {k: v, …}}ctx["fields"]
confirm_surface{requestId, confirmed: bool}ctx["confirmed"]

Hand-built surfaces (surface_request(components=[...])) set whatever context they want; the agent reads the matching key.

Example — picker:

def _router(state):
    hitl = state.get("hitl_response")
    if not hitl:
        return {}  # turn 1 — no HITL yet
    ctx = (hitl.get("action") or {}).get("context") or {}
    selected = ctx.get("selected") or []   # this is what the user picked
    if not selected:
        return {"cancelled": True}
    questions = state.get("research_questions", [])
    filtered = [questions[int(i)] for i in selected if str(i).isdigit()]
    return {"research_questions": filtered, "current_question_index": 0}

Chat-transcript note. The platform ALSO injects a human-readable user-message body into the conversation (rendered by buildHitlResumeMessage on the server). That body is for chat-log display only — an agent must not parse it. Read hitl_response instead.

What the platform does NOT do. It does not translate between v0.9 submit-context keys (selected, value, fields, confirmed) and legacy payload keys (selectedIds, parameters, approved). Agents written against pre-v0.9 envelope field names need to update to the contract above. The /api/hitl/:id/respond endpoint remains available as a transitional helper for non-Portal clients but is deprecated; new integrations should use POST /api/surfaces/:id/resolve with the canonical {action, dataModel} body.

Invocation Paths: End-User Chat vs. Agent-to-Agent Delegation

An SDK agent can be invoked two ways. The SDK transparently handles both, but the runtime semantics differ:

Entry pointCallerWho runs the tool-execution loop?Agent response shape
POST /ag-uiPlatform's external-agent Temporal workflow (end-user chat in Portal)Platform — intercepts TASK_STATE_WORKING, runs each declared tool as a Temporal activity (grants, HITL, retry, audit), re-calls /ag-ui with role=tool messagesCan pause with TASK_STATE_WORKING + governed-tool-call parts between segments
POST / (JSON-RPC tasks.send)Another agent (A2A delegation) or external clientAgent — self-executes declared tool calls via Platform MCP inside on_a2a_message()Only terminal states: TASK_STATE_COMPLETED or TASK_STATE_INPUT_REQUIRED

The SDK's server.py sets context["_ag_ui_invocation"] = True on the AG-UI entry point. The default on_a2a_message() implementation branches on this flag: AG-UI path emits declared tool calls and stops for the platform's loop; A2A path runs an internal loop over PlatformMCPClient and returns only terminal task states. This is mandatory — the platform's a2a-invoker treats a TASK_STATE_WORKING response on the A2A path as an incomplete delegation and raises an explicit error (no silent fallback). An A2A-invoked sub-agent that returns early without executing its tools will never produce a complete answer.

Governance still flows on both paths. On the A2A path, server.py parses the X-Delegation-Token, X-Effective-Tool-Grants, and X-Delegation-Chain headers and stashes them on the agent so the internal loop's PlatformMCPClient re-emits them on every outbound MCP call. The MCP server enforces the delegator ∩ delegatee intersection at tool-execution time, so a sub-agent can never exceed the grants its caller had.

Distributed Tracing (W3C Trace Context)

The platform propagates W3C traceparent and tracestate headers on every outbound call — AG-UI invocations, A2A delegations, and MCP tool calls. The SDK captures incoming values at both /ag-ui and / entry points and re-emits them on outbound MCP calls via PlatformMCPClient. Agents do not need OpenTelemetry installed; the SDK forwards the headers verbatim. Result: the entire request tree (user → orchestrator agent → sub-agent → tool) shows up as a single trace in the OTel backend, with per-agent spans nested under the root user turn. No call-site changes are needed in agent code.

Agent Lifecycle

  provision     dev           test         register        deploy
  ─────> Clone ────────> Develop ─────> Validate ─────> Register ─────> Deploy
         repo            locally         contract        metadata       container
  1. Provision / clone — the platform provisions your repo (rendered from customer/template/); you git clone it. Repos are no longer self-scaffolded.
  2. agentic dev — Start local server with hot-reload
  3. agentic test — Validate AG-UI protocol compliance
  4. agentic register — Register agent metadata + tool grants + skills
  5. agentic deploy — Build Docker image, push, and start container

Python SDK

Installation

pip install agentic-hub-sdk

Basic Agent

from agentic_sdk import (
    AgenticAgent,
    AgentTurnRequest,
    AgentTurnResponse,
    message,
    done,
    run,
)

class MyAgent(AgenticAgent):
    name = "my-agent"
    description = "Echoes user messages"
    version = "1.0.0"

    async def on_turn(self, request: AgentTurnRequest) -> AgentTurnResponse:
        user_msg = next(
            (m.content for m in reversed(request.messages) if m.role == "user"),
            "Hello",
        )
        return done(message("assistant", f"Echo: {user_msg}"))

if __name__ == "__main__":
    run(MyAgent(), port=8000)

Using Tools

from agentic_sdk import (
    AgenticAgent, AgentTurnRequest, AgentTurnResponse,
    message, done, request_tools, tool_call,
)

class ResearchAgent(AgenticAgent):
    name = "research-agent"
    capabilities = ["research", "summarize"]

    async def on_turn(self, request: AgentTurnRequest) -> AgentTurnResponse:
        if request.turn_number == 0:
            user_msg = next(
                m.content for m in reversed(request.messages) if m.role == "user"
            )
            return request_tools(
                message("assistant", f"Searching for: {user_msg}"),
                tool_calls=[tool_call("web_search", {"query": user_msg})],
            )

        results = request.tool_results[0].output if request.tool_results else "No results"
        return done(message("assistant", f"Found: {results}"))

Agent API

Method / PropertyDescription
on_turn(request)Required. Process a turn and return a response.
on_init()Called once at startup. Override for initialization.
on_shutdown()Called at shutdown. Override for cleanup.
on_health()Health check. Override for custom logic.
on_error(error, request)Called when on_turn raises. Override for custom error handling.
get_tool(name)Look up an available tool by name.
has_tool(name)Check if a tool is available.
available_toolsList of tools available to this agent.
stateCurrent turn state dict (from context).

Helper Functions

FunctionDescription
message(role, content)Create an AgentMessage
tool_call(name, args)Create an AgentToolCall
done(message)Create response with next_action="done"
request_tools(message, tool_calls=[], state_patch={})Create response requesting tool execution
await_approval(message, state_patch={})Pause for content-level HITL (plan selection, parameter edit, user input)
request_approval(hitl_type, options, message)Build an A2UI surface and pause for user approval

Content-Level HITL via A2UI Protocol

The SDK uses the A2UI protocol (v0.8) to present declarative approval surfaces to the user. A2UI defines a standard vocabulary of UI components (selection lists, parameter forms, free-text inputs) that any compliant frontend can render — no custom widget code required.

Using request_approval() (recommended)

The request_approval() helper builds a standard A2UI v0.8 surface automatically from your HITL type and options:

from agentic_sdk import request_approval, message

# Plan selection — renders as an A2UI selectable list
return request_approval(
    hitl_type="plan_selection",
    options=[{"id": "0", "title": "Option A"}, {"id": "1", "title": "Option B"}],
    message=message("assistant", "Here is my plan — please select which options to proceed with."),
)

# Parameter edit — renders as an A2UI editable form
return request_approval(
    hitl_type="parameter_edit",
    options={"max_retries": 3, "timeout_seconds": 30},
    message=message("assistant", "Please review and adjust these parameters."),
)

# Free-text user input — renders as an A2UI text input
return request_approval(
    hitl_type="user_input",
    message=message("assistant", "What additional context should I consider?"),
)

Under the hood, request_approval() returns next_action="await_approval" with the appropriate state_patch containing A2UI surface metadata. The SDK emits a RUN_FINISHED AG-UI event with the governance extension namespace (urn:agentic-platform:governance:v1) to signal that the agent is paused awaiting human input.

Using await_approval() (low-level)

For advanced cases, you can construct the state patch directly:

return {
    "messages": [message("assistant", "Here is my plan...")],
    "next_action": "await_approval",
    "state_patch": {
        "hitl_type": "plan_selection",
        "hitl_options": [{"id": "0", "title": "Option A"}, {"id": "1", "title": "Option B"}],
    },
}

How HITL Approval Works

When an agent emits a surface via surface_request(...), the platform:

  1. Persists a surface_requests row keyed off _meta.surfaceRequestId.
  2. Writes a surface:emit audit event (dimension=2).
  3. Streams the A2UI envelopes (createSurface, updateComponents, updateDataModel) to the client.
  4. Pauses the Temporal workflow via condition() and waits for the canonical resume signal surfaceResolve.

When the user submits the surface, the client (Portal / SDK consumer) POSTs the canonical v0.9 envelope to POST /api/surfaces/:surfaceRequestId/resolve:

{
  "action": {
    "kind": "event",
    "name": "urn:a2ui:submit",
    "surfaceId": "hitl-<requestId>",
    "context": { "requestId": "<requestId>", "selected": ["0", "2"] }
  },
  "dataModel": null
}

The platform writes a surface:resolve audit event (dimension=2, context_keys=[...] keys-only) and signals Temporal. The agent receives the envelope verbatim as hitl_response in its context, and reads its own submit payload via hitl_response.action.context.<field> — see §"HITL Response Contract" above for the shape matrix. There is no legacy type-tag (hitl_plan_selection_response etc.) translation on any path. Token usage is tracked automatically across all HITL pauses.

Protocol Integration

The HITL flow uses three standard protocols working together:

  • A2A (Agent-to-Agent): Manages the agent lifecycle and turn-based execution. The RUN_FINISHED event carries a governance extension (urn:agentic-platform:governance:v1) indicating HITL status.
  • AG-UI (Agent-to-UI pipe): Streams execution events (text deltas, tool calls, state updates) to the frontend in real time.
  • A2UI (Agent-to-UI content): Defines the declarative surface schema for approval UIs. The SDK builds standard A2UI v0.8 components (selection lists, parameter forms, text inputs) that any compliant renderer can display.

TypeScript SDK

Installation

npm install @agentic-platform/sdk

Basic Agent

import { AgenticAgent, AgentTurnRequest, AgentTurnResponse, msg, done, serve } from '@agentic-platform/sdk';

class MyAgent extends AgenticAgent {
  name = 'my-agent';
  description = 'Echoes user messages';

  async onTurn(request: AgentTurnRequest): Promise<AgentTurnResponse> {
    const userMsg = [...request.messages].reverse().find(m => m.role === 'user')?.content ?? '';
    return done(msg('assistant', `Echo: ${userMsg}`));
  }
}

serve(new MyAgent(), { port: 8000 });

Server Factory

For more control over the HTTP server (add middleware, custom routes):

import { createServer } from '@agentic-platform/sdk';

const app = createServer(new MyAgent());
// app is a Hono instance — add custom routes, middleware, etc.
export default { port: 8000, fetch: app.fetch };

Using Tools (Multi-Segment Flow)

The TypeScript SDK handles the multi-segment AG-UI flow automatically. When an agent returns toolCalls, the platform executes them and calls the agent again with tool results:

import { AgenticAgent, AgentTurnRequest, AgentTurnResponse, msg, done } from '@agentic-platform/sdk';
import type { ToolRequirement } from '@agentic-platform/sdk';

class ResearchAgent extends AgenticAgent {
  name = 'research-agent';
  requiredTools: ToolRequirement[] = [
    { toolId: 'tool-web-search', name: 'web_search' },
  ];

  async onTurn(request: AgentTurnRequest): Promise<AgentTurnResponse> {
    const phase = (request.context?.phase as string) || 'initial';

    if (phase === 'initial') {
      const userMsg = [...request.messages].reverse().find(m => m.role === 'user')?.content ?? '';
      return {
        messages: [msg('assistant', `Searching for: ${userMsg}`)],
        toolCalls: [{ id: `search-${Date.now()}`, name: 'web_search', arguments: { query: userMsg } }],
        nextAction: 'await_tool_results',
        statePatch: { phase: 'awaiting_results', query: userMsg },
      };
    }

    if (phase === 'awaiting_results') {
      const results = request.toolResults?.map(tr => tr.output).join('\n') || 'No results';
      return done(msg('assistant', `Found: ${results}`));
    }

    return done(msg('assistant', 'How can I help?'));
  }
}

Key points:

  • turnNumber is set automatically: 0 for initial turns, > 0 when tool results are present
  • request.toolResults contains the results from the previous turn's toolCalls
  • statePatch is preserved between segments via STATE_SNAPSHOT events (emitted automatically by the SDK)
  • request.context contains the statePatch from the previous segment

Content-Level HITL via A2UI (TypeScript)

The TypeScript SDK has full A2UI surface builder parity with the Python SDK:

import {
  buildApprovalSurface,
  buildPlanSelectorSurface,
  buildParameterEditorSurface,
  buildUserInputSurface,
  buildDeleteSurface,
  buildA2UIFromHITL,
} from '@agentic-platform/sdk';

// Approval surface — renders as Card with Approve/Reject buttons
const surfaces = buildApprovalSurface('req-1', 'create_ticket', { title: 'Bug fix' }, 0.5, 'Creating ticket');

// Plan selection — renders as MultipleChoice list
const surfaces = buildPlanSelectorSurface('req-2', 'Select a plan:', [
  { id: '0', title: 'Option A', description: 'Fast but costly' },
  { id: '1', title: 'Option B', description: 'Slow but cheap' },
]);

// Parameter editor — renders as editable form
const surfaces = buildParameterEditorSurface('req-3', 'Review parameters:', 'deploy', [
  { name: 'replicas', type: 'number', value: 3, label: 'Replica count' },
  { name: 'region', type: 'string', value: 'eu-west', label: 'Region' },
]);

// Free-text input — renders as TextField
const surfaces = buildUserInputSurface('req-4', 'Additional context?', 'textarea', 'Enter details...');

Automatic A2UI emission: When an agent returns nextAction: 'await_approval' with HITL metadata in statePatch, the SDK automatically:

  1. Emits a STATE_SNAPSHOT event to preserve state between segments
  2. Calls buildA2UIFromHITL(statePatch, runId) to generate A2UI surfaces
  3. Emits them as CUSTOM events (AG-UI path) or DataPart entries (A2A path)
// In your agent's onTurn():
return {
  messages: [msg('assistant', 'Shall I create the ticket?')],
  nextAction: 'await_approval',
  statePatch: {
    hitl_type: 'approval',
    hitl_tool_name: 'create_ticket',
    hitl_tool_args: { title: 'Bug fix', body: '...' },
    hitl_reasoning: 'Creating a ticket for the reported issue',
    request_id: `approval-${Date.now()}`,
  },
};

Declaring Tools

Referencing Existing Tools

Use ToolRequirement to declare tools that already exist on the platform:

Python:

from agentic_sdk import AgenticAgent, ToolRequirement

class MyAgent(AgenticAgent):
    name = "my-agent"
    required_tools = [
        ToolRequirement(
            tool_id="tool-kb",
            permissions=["read", "execute"],
        ),
        ToolRequirement(
            tool_id="tool-web-search",
            permissions=["read", "execute"],
        ),
        ToolRequirement(
            tool_id="tool-ticket",
            permissions=["read", "write", "execute"],
            conditions={"requireApproval": True, "approvalPolicy": "hitl-manager"},
        ),
    ]

TypeScript:

import { AgenticAgent, ToolRequirement } from '@agentic-platform/sdk';

class MyAgent extends AgenticAgent {
  name = 'my-agent';
  requiredTools: ToolRequirement[] = [
    { toolId: 'tool-kb', permissions: ['read', 'execute'] },
    { toolId: 'tool-web-search', permissions: ['read', 'execute'] },
  ];
}

When you run agentic register, the CLI reads these declarations from the agent card and automatically creates tool grants on the platform.

Creating New Tools

Use ToolDefinition to ship custom MCP tools with your agent:

from agentic_sdk import AgenticAgent, ToolDefinition

class MyAgent(AgenticAgent):
    name = "my-agent"
    new_tools = [
        ToolDefinition(
            name="Log Parser",
            mcp_endpoint="mcp://my-server/parse-logs",
            description="Parse and analyze application log files",
            risk_level="low",
            category="read-only",
            input_schema={
                "type": "object",
                "properties": {
                    "logFile": {"type": "string"},
                    "timeRange": {"type": "string"},
                },
            },
        ),
    ]

The agentic register command will:

  1. Create the tool on the platform via POST /api/tools
  2. Automatically grant the tool to your agent

Declaring Skills

Skills are curated instruction bundles (runbooks, procedures, guidelines) that guide agent behavior.

Referencing Existing Skills

from agentic_sdk import AgenticAgent, SkillRequirement

class MyAgent(AgenticAgent):
    name = "my-agent"
    skills = [
        SkillRequirement(skill_id="skill-code-review"),
        SkillRequirement(skill_id="skill-system-ops"),
    ]

Creating New Skills

from agentic_sdk import AgenticAgent, SkillDefinition

class MyAgent(AgenticAgent):
    name = "my-agent"
    new_skills = [
        SkillDefinition(
            name="incident-triage",
            instructions="""# Incident Triage Procedure

1. **Assess Severity**: Check monitoring dashboards
2. **Identify Scope**: Determine affected services
3. **Communicate**: Post status update
4. **Mitigate**: Apply immediate fix
5. **Document**: Create post-mortem
""",
            category="operations",
            tags=["sre", "incident", "triage"],
            visibility="tenant",
        ),
    ]

Framework Adapters

The Python SDK includes adapters to wrap existing agent frameworks for platform compatibility.

PydanticAI

from agentic_sdk import AgenticAgent, AgentTurnRequest, AgentTurnResponse
from agentic_sdk.adapters.pydantic_ai import PydanticAIAdapter
from pydantic_ai import Agent

pydantic_agent = Agent("openai:gpt-4.1-mini", system_prompt="You are helpful.")

class MyAgent(AgenticAgent):
    name = "pydantic-agent"

    async def on_turn(self, request: AgentTurnRequest) -> AgentTurnResponse:
        adapter = PydanticAIAdapter(pydantic_agent)
        return await adapter.handle_turn(request)

CrewAI

from agentic_sdk.adapters.crewai import CrewAIAdapter
from crewai import Agent, Task, Crew

crew = Crew(agents=[...], tasks=[...])

class MyAgent(AgenticAgent):
    name = "crew-agent"

    async def on_turn(self, request: AgentTurnRequest) -> AgentTurnResponse:
        adapter = CrewAIAdapter(crew)
        return await adapter.handle_turn(request)

AutoGen

from agentic_sdk.adapters.autogen import AutoGenAdapter

class MyAgent(AgenticAgent):
    name = "autogen-agent"

    async def on_turn(self, request: AgentTurnRequest) -> AgentTurnResponse:
        adapter = AutoGenAdapter(agent_config={...})
        return await adapter.handle_turn(request)

LangGraph

from agentic_sdk.adapters.langgraph import LangGraphAdapter
from langgraph.graph import StateGraph, END
from typing import TypedDict, Any

class MyState(TypedDict):
    messages: list[dict[str, str]]
    research_questions: list[str]
    current_question_index: int
    tool_results: list[dict[str, Any]]
    tool_calls: list[dict[str, Any]]
    next_action: str
    state_patch: dict[str, Any]
    _token_usage: dict[str, int]
    # IMPORTANT: Any custom flags you need across turns MUST be in the TypedDict.
    # LangGraph drops keys not in the schema from the output state, so the
    # LangGraphAdapter's auto-capture won't preserve them between segments.
    _my_custom_flag: bool

builder = StateGraph(MyState)
# ... add nodes and edges ...
compiled = builder.compile()

adapter = LangGraphAdapter(compiled, name="my-agent", supports_a2a=True)

Multi-turn state management: The LangGraphAdapter automatically captures non-protocol fields from the graph output into state_patch, which the platform carries forward between AG-UI segments. Protocol fields (messages, tool_calls, next_action, state_patch, tool_results, turn_number, _token_usage) are excluded from auto-capture.

⚠️ Important: Any custom state keys you need to persist across turns (e.g., flags, counters) must be declared in your TypedDict. LangGraph silently drops undeclared keys from the output state, preventing auto-capture.

AG-UI Protocol

The AG-UI adapter adds a streaming endpoint for AG-UI compatible clients:

from agentic_sdk import create_server
from agentic_sdk.adapters.agui import mount_agui_endpoint

app = create_server(MyAgent())
mount_agui_endpoint(app, MyAgent())
# Adds POST /agui/events endpoint

Testing

MockPlatform (In-Process Testing)

Test your agent without running a server:

import pytest
from agentic_sdk.testing import MockPlatform

@pytest.mark.asyncio
async def test_echo():
    platform = MockPlatform(MyAgent())
    response = await platform.send_message("Hello")
    assert "Hello" in response

@pytest.mark.asyncio
async def test_with_tools():
    platform = MockPlatform(MyAgent())
    platform.register_tool(
        "web_search",
        lambda args: {"results": [{"title": "Test", "url": "https://example.com"}]},
    )
    response = await platform.send_message("Search for Python")
    assert "Test" in response

ContractTestSuite (HTTP Testing)

Validate that a running agent correctly implements the AG-UI protocol:

from agentic_sdk.testing import ContractTestSuite

suite = ContractTestSuite("http://localhost:8000")
results = await suite.run_all()
for result in results:
    print(f"{'PASS' if result.passed else 'FAIL'} {result.name} ({result.duration_ms}ms)")

The suite tests:

  • Health endpoint (GET /health)
  • Agent card (GET /.well-known/agent.json)
  • AG-UI message processing (POST /ag-ui)
  • Multi-turn conversations
  • Error handling

TypeScript Testing

import { MockPlatform, ContractTestSuite } from '@agentic-platform/sdk/testing';

// In-process testing
const platform = new MockPlatform(new MyAgent());
platform.registerTool('web_search', (args) => ({ results: [] }));
const response = await platform.sendMessage('Hello');

// HTTP contract testing
const suite = new ContractTestSuite('http://localhost:8000');
const results = await suite.runAll();

Platform Client

The Python SDK includes a PlatformClient for programmatic platform interaction:

from agentic_sdk import PlatformClient

async with PlatformClient("http://localhost:4000", token="my-token") as client:
    # Register an agent
    agent = await client.register_agent({
        "name": "my-agent",
        "runtime": "procode",
        "executionMode": "external-procode",
    })

    # List agents
    agents = await client.list_agents()

    # Create a tool
    tool = await client.create_tool({
        "name": "My Parser",
        "mcpEndpoint": "mcp://server/parse",
    })

    # Grant a tool to an agent
    await client.grant_tool(
        tool_id=tool["id"],
        agent_id=agent["id"],
        permissions=["read", "execute"],
    )

    # Create and assign a skill
    skill = await client.create_skill({
        "name": "triage",
        "instructions": "Step 1: ...",
    })
    await client.assign_skill(agent_id=agent["id"], skill_id=skill["id"])

Agent Configuration: Pipeline & UX Options

The AgentConfig type supports several fields for deployment pipeline configuration and user experience.

Suggested Prompts

Define conversation starters shown in the Portal chat UI:

class MyAgent(AgenticAgent):
    name = "my-agent"
    config = {
        "suggestedPrompts": [
            "What can you help me with?",
            "Summarize my latest project status",
            "Find open issues assigned to me",
        ],
    }

If suggestedPrompts is not set, the Portal auto-generates prompts from the agent's capabilities.

Deployment Pipeline Configuration

Control which deployment stages run and their thresholds via pipelineConfig:

class MyAgent(AgenticAgent):
    name = "my-agent"
    config = {
        "pipelineConfig": {
            "evalSuiteId": "eval-core-qa",       # ID of the eval suite to run
            "evalThreshold": 85,                   # Minimum pass rate (0-100)
            "canaryPercent": 10,                   # Canary traffic percentage
            "autoRollbackOnFailure": True,         # Auto-rollback on stage failure
            "skipStages": ["behavior_check"],      # Stages to skip
        },
    }

The platform also tracks versionHistory automatically on each deployment, recording the version, timestamp, and deployer.

Credential Resolution Preview

The platform provides a dry-run endpoint to inspect which tier a credential resolves from without exposing its value:

async with PlatformClient("http://localhost:4000", token="my-token") as client:
    preview = await client.get("/api/credentials/resolve-preview", params={
        "credentialKey": "OPENAI_API_KEY",
        "agentId": "agent-research-1",
    })
    # Returns: { tier: 2, source: "agent-override", masked: "sk-...xxxx" }

This is useful for debugging credential resolution across the 5-tier hierarchy (Vault, User BYOK, Agent override, Tenant default, Environment variable).

Deployment

Docker Base Images

Pre-configured images with OTel tracing, health checks, and the SDK pre-installed:

Python Agent:

FROM ghcr.io/agentic-platform/python-agent:0.1.0
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "agent.py"]

Node.js Agent:

FROM ghcr.io/agentic-platform/node-agent:0.1.0
COPY package.json .
RUN npm install --production
COPY . .
CMD ["node", "dist/agent.js"]

Deployment Flow

# Build and push Docker image
agentic deploy --registry ghcr.io/my-org

# Monitor
agentic status
agentic logs --follow

The agentic deploy command:

  1. Reads agentic.yaml for configuration
  2. Builds a Docker image using the project's Dockerfile
  3. Pushes to the specified registry
  4. Registers or updates the agent on the platform
  5. Starts the container via the lifecycle API
  6. Polls until the container is running

Risk Classification

Set the risk level to control governance requirements:

LevelDescriptionRequirements
assistiveRead-only, low-risk actionsStandard guardrails
semi-autonomousCan modify data with oversightHITL for high-risk actions
autonomousFull autonomy within granted scopeComprehensive audit trail
class MyAgent(AgenticAgent):
    name = "my-agent"
    risk_classification = "semi-autonomous"

CLI Reference

See the CLI README for full command reference.

Example: SRE Incident Response Agent

A complete example is available at examples/sre-incident-agent/. This agent demonstrates:

  • 6 tool declarations (knowledge base, web search, code execution, database, ticketing with HITL, Slack)
  • 1 skill assignment (system operations)
  • Multi-turn incident triage workflow
  • Semi-autonomous risk classification
  • 16 tests covering the full agent lifecycle
cd examples/sre-incident-agent
pip install -r requirements.txt
python agent.py                    # Start on port 8400
agentic test --url http://localhost:8400   # Validate contract
agentic register                   # Register with platform

Example: Node.js Web Search Agent with HITL

A complete TypeScript SDK example is available at examples/agent-examples/nodejs_agent.ts. This agent demonstrates:

  • Multi-phase execution: search → summarize → approve → create ticket
  • Web search tool invocation via governed tool calls
  • HITL approval flow with A2UI rich surfaces (Approve/Reject card)
  • Automatic STATE_SNAPSHOT emission for multi-segment Temporal workflows
  • Docker deployment via Dockerfile.nodejs
# Local development
cd examples/agent-examples
npx tsx nodejs_agent.ts          # Start on port 8304

# Docker deployment (managed by platform)
docker compose build nodejs-web-search-agent
docker compose up nodejs-web-search-agent

The agent is pre-registered in the platform seed data as agent-nodejs-web-search with tool grants for web_search and create_ticket (with HITL approval).

Further Reading