Stacksona icon
Stacksona Agent Governance Protocol
PydanticAI Integration Blueprint

Framework runs the agent, Stacksona handles the gate via API calls.

Stacksona tracks deferred tool lifecycle events and ties decisions to typed output validation paths.

Governance object: deferred tool request + typed output validation.

Official docs ↗Deferred tools ↗Output API ↗

Sequence Diagram

Core workflow endpoints in this flow: POST /v1/tasks → POST /v1/tasks/{task_id}/events → POST /v1/gate/decision.

Why this gate point: PydanticAI governance preserves typed guarantees across deferred side effects and validator stages.

Technical Spec

Decision API: POST /v1/gate/decision with response fields decision, reviewer_id, and audit_hash.
TASK CREATE API · POST /v1/tasks
{
  "required": ["agent_id", "framework_origin", "task_name"],
  "properties": {
    "agent_id": "string",
    "framework_origin": "LangGraph|CrewAI|AutoGen|LlamaIndex|PydanticAI|CustomAPI|Other",
    "task_name": "string",
    "task_description": "string",
    "governance_context": {},
    "metadata": {}
  }
}
TASK EVENT API · POST /v1/tasks/{task_id}/events
{
  "required": ["task_id", "agent_id", "event_type", "event_payload", "framework_origin", "occurred_at", "correlation_id"],
  "event_type": [
    "TASK_STARTED", "TASK_UPDATED", "TOOL_REQUESTED", "TOOL_EXECUTED",
    "GUARDRAIL_TRIGGERED", "REVIEW_REQUESTED", "REVIEW_RESULT_RECEIVED",
    "TASK_COMPLETED", "TASK_FAILED", "TASK_REVISED"
  ],
  "properties": {
    "task_id": "string",
    "agent_id": "string",
    "framework_origin": "LangGraph|CrewAI|AutoGen|LlamaIndex|PydanticAI|CustomAPI|Other",
    "occurred_at": "date-time",
    "correlation_id": "string",
    "event_payload": {},
    "metadata": {}
  }
}
PYTHON HOOKS · TASK CREATE + EVENT INGEST
import requests

def create_task(api_key, payload, base_url="https://api.stacksona.com"):
    url = f"{base_url}/v1/tasks"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    response = requests.post(url, json=payload, headers=headers, timeout=30)
    response.raise_for_status()
    return response.json()

def send_task_event(api_key, task_id, payload, base_url="https://api.stacksona.com"):
    url = f"{base_url}/v1/tasks/{task_id}/events"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    response = requests.post(url, json=payload, headers=headers, timeout=30)
    response.raise_for_status()
    return response.json()

This is the task lifecycle logging path. Runtime creates a task via POST /v1/tasks, logs timeline events via POST /v1/tasks/{task_id}/events, then requests and consumes governance decisions with full audit correlation.

Implementation Patterns

Task-first Correlation

Create task_id before governance so all runtime events share one audit lineage.

Event-driven Audit Trail

Send every task milestone through POST /v1/tasks/{task_id}/events for immutable timeline visibility.

Decision-linked Continuation

Apply APPROVED/DENIED/REVISE decisions and continue framework-native flow with reviewer and audit hash context.

Speak with a Governance EngineerView Full API Reference