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 Dual-Protocol Model (AG-UI + A2A)
- Agent Lifecycle
- Python SDK
- TypeScript SDK
- Declaring Tools
- Declaring Skills
- Framework Adapters
- Testing
- Platform Client
- Deployment
- CLI Reference
- Example: SRE Incident Response Agent
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 responseTOOL_CALL_START/TOOL_CALL_ARGS/TOOL_CALL_END— Declare tool callsSTATE_SNAPSHOT/STATE_DELTA— Share agent stateRUN_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:
| Owner | What |
|---|---|
| Platform | task.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 turn | Exactly 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_patch | Agent-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 — private | Internal 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 helper | Submit context shape | Agent 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 point | Caller | Who runs the tool-execution loop? | Agent response shape |
|---|---|---|---|
POST /ag-ui | Platform'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 messages | Can pause with TASK_STATE_WORKING + governed-tool-call parts between segments |
POST / (JSON-RPC tasks.send) | Another agent (A2A delegation) or external client | Agent — 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
- Provision / clone — the platform provisions your repo (rendered from
customer/template/); yougit cloneit. Repos are no longer self-scaffolded. agentic dev— Start local server with hot-reloadagentic test— Validate AG-UI protocol complianceagentic register— Register agent metadata + tool grants + skillsagentic 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 / Property | Description |
|---|---|
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_tools | List of tools available to this agent. |
state | Current turn state dict (from context). |
Helper Functions
| Function | Description |
|---|---|
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:
- Persists a
surface_requestsrow keyed off_meta.surfaceRequestId. - Writes a
surface:emitaudit event (dimension=2). - Streams the A2UI envelopes (
createSurface,updateComponents,updateDataModel) to the client. - Pauses the Temporal workflow via
condition()and waits for the canonical resume signalsurfaceResolve.
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_FINISHEDevent 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:
turnNumberis set automatically:0for initial turns,> 0when tool results are presentrequest.toolResultscontains the results from the previous turn'stoolCallsstatePatchis preserved between segments viaSTATE_SNAPSHOTevents (emitted automatically by the SDK)request.contextcontains thestatePatchfrom 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:
- Emits a
STATE_SNAPSHOTevent to preserve state between segments - Calls
buildA2UIFromHITL(statePatch, runId)to generate A2UI surfaces - Emits them as
CUSTOMevents (AG-UI path) orDataPartentries (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:
- Create the tool on the platform via
POST /api/tools - 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:
- Reads
agentic.yamlfor configuration - Builds a Docker image using the project's Dockerfile
- Pushes to the specified registry
- Registers or updates the agent on the platform
- Starts the container via the lifecycle API
- Polls until the container is running
Risk Classification
Set the risk level to control governance requirements:
| Level | Description | Requirements |
|---|---|---|
assistive | Read-only, low-risk actions | Standard guardrails |
semi-autonomous | Can modify data with oversight | HITL for high-risk actions |
autonomous | Full autonomy within granted scope | Comprehensive 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_SNAPSHOTemission 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).