Stacksona icon
Stacksona Agent Governance Protocol
LlamaIndex Integration Blueprint

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

Stacksona logs retrieval events and review decisions to task timeline before final response synthesis.

Governance object: retrieved nodes + citation/source policy.

Official docs ↗Retriever guide ↗Response synthesizers ↗

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: RAG governance is clearest between retrieval and synthesis where source quality and citation completeness are known.

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