API Overview - Shannon

Introduction

Shannon provides HTTP REST and gRPC APIs for submitting tasks, streaming results, and managing AI agent workflows. Base URL: http://localhost:8080 (development)

Quick Start

# Submit a task
curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{"query": "Analyze sentiment: Shannon makes AI simple"}'

# Response
{
  "task_id": "task-dev-1234567890",
  "status": "submitted",
  "created_at": "2025-01-20T10:00:00Z"
}

# Stream real-time events
curl -N "http://localhost:8080/api/v1/stream/sse?workflow_id=task-dev-1234567890"

API Endpoints

Core Task Operations

Endpoint	Method	Description
`/api/v1/tasks`	POST	Submit a new task
`/api/v1/tasks/stream`	POST	Submit task and get stream URL (201)
`/api/v1/tasks`	GET	List tasks with filters
`/api/v1/tasks/{id}`	GET	Get task status and result
`/api/v1/tasks/{id}/stream`	GET	Stream events for a specific task
`/api/v1/tasks/{id}/events`	GET	Get task event history
`/api/v1/tasks/{id}/timeline`	GET	Get Temporal execution timeline

Task Control

Endpoint	Method	Description
`/api/v1/tasks/{id}/cancel`	POST	Cancel a running task
`/api/v1/tasks/{id}/pause`	POST	Pause task at next checkpoint
`/api/v1/tasks/{id}/resume`	POST	Resume a paused task
`/api/v1/tasks/{id}/control-state`	GET	Get task control state (paused/running)

Real-Time Streaming

Endpoint	Method	Description
`/api/v1/stream/sse?workflow_id={id}`	GET	Server-Sent Events stream
`/api/v1/stream/ws?workflow_id={id}`	GET	WebSocket stream

Session Management

Endpoint	Method	Description
`/api/v1/sessions`	GET	List sessions with pagination
`/api/v1/sessions/{sessionId}`	GET	Get session details
`/api/v1/sessions/{sessionId}/history`	GET	Get session chat history
`/api/v1/sessions/{sessionId}/events`	GET	Get session events (grouped by turn)
`/api/v1/sessions/{sessionId}`	PATCH	Update session title (UUID or external_id)
`/api/v1/sessions/{sessionId}`	DELETE	Delete session (soft delete; 204 idempotent)

Schedules (Recurring Tasks)

Endpoint	Method	Description
`/api/v1/schedules`	POST	Create a new schedule
`/api/v1/schedules`	GET	List schedules with filters
`/api/v1/schedules/{id}`	GET	Get schedule details
`/api/v1/schedules/{id}/runs`	GET	Get schedule execution history
`/api/v1/schedules/{id}`	PUT	Update schedule configuration
`/api/v1/schedules/{id}/pause`	POST	Pause a schedule
`/api/v1/schedules/{id}/resume`	POST	Resume a paused schedule
`/api/v1/schedules/{id}`	DELETE	Delete a schedule

Authentication

Endpoint	Method	Description	Auth Required
`/api/v1/auth/register`	POST	Register a new user	No
`/api/v1/auth/login`	POST	Login and get JWT tokens	No
`/api/v1/auth/refresh`	POST	Refresh JWT access token	No
`/api/v1/auth/me`	GET	Get current user info	Yes
`/api/v1/auth/refresh-key`	POST	Regenerate API key	Yes
`/api/v1/auth/api-keys`	GET	List user’s API keys	Yes
`/api/v1/auth/api-keys`	POST	Create new API key	Yes
`/api/v1/auth/api-keys/{id}`	DELETE	Revoke an API key	Yes

Approvals

Endpoint	Method	Description
`/api/v1/approvals/decision`	POST	Submit approval decision for human-in-the-loop tasks

OpenAI-Compatible API

Shannon provides an OpenAI-compatible API for easy integration with existing OpenAI SDKs and tools. See the OpenAI-Compatible API Reference for full details.

Endpoint	Method	Description
`/v1/chat/completions`	POST	Chat completions (streaming supported)
`/v1/models`	GET	List available models
`/v1/models/{model}`	GET	Get specific model details

Health & Observability

Endpoint	Method	Description	Auth Required
`/health`	GET	Health check	No
`/readiness`	GET	Readiness probe	No
`/openapi.json`	GET	OpenAPI 3.0 spec	No

Authentication

Development Default: Authentication is disabled (GATEWAY_SKIP_AUTH=1). Enable for production.

When enabled, pass API key via header:

curl -H "X-API-Key: sk_test_123456" \
  http://localhost:8080/api/v1/tasks

SSE Streaming: Browser EventSource cannot send custom headers.

Development: set GATEWAY_SKIP_AUTH=1 to skip auth.
Production: initiate SSE from your backend (or edge) and inject X-API-Key or Authorization: Bearer headers.
For SSE endpoints, the api_key query parameter is supported as a fallback when headers cannot be sent.

Response Format

All endpoints return JSON with consistent error format: Success (200):

{
  "task_id": "task-dev-1234567890",
  "status": "COMPLETED",
  "result": "Sentiment: Positive"
}

Error (400/401/404/429/500):

{
  "error": "Task not found"
}

Rate Limiting

Default: 60 requests/minute per API key
Headers: X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset
429 Response: Includes Retry-After header

Event Streaming

Shannon emits 33 event types for real-time monitoring: Core Events:

WORKFLOW_STARTED, WORKFLOW_COMPLETED
AGENT_STARTED, AGENT_COMPLETED
STATUS_UPDATE, DATA_PROCESSING
ERROR_OCCURRED

LLM Events:

LLM_PROMPT, LLM_OUTPUT, LLM_PARTIAL

Tool Events:

TOOL_INVOKED, TOOL_OBSERVATION

See Event Types Reference for the complete list. Python SDK: use the EventType enum for filtering and checks. See SDK Streaming.

Client Access

Use the Gateway REST API or the Python SDK (HTTP-only):

REST endpoints: http://localhost:8080/api/v1/*
Python SDK: ShannonClient(base_url="http://localhost:8080")

Note: gRPC services are internal and not part of the public SDK surface. The OpenAI-compatible /v1/chat/completions endpoint is intended for compatibility. For full Shannon features (skills, session workspaces, research strategies), use /api/v1/tasks and related endpoints.

Best Practices

1. Use Session IDs for Context

{
  "query": "What about Q2?",
  "session_id": "user-123-analytics"
}

2. Stream Events for Long Tasks

import requests

r = requests.get(
    f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}",
    stream=True
)

for line in r.iter_lines():
    if line:
        event = parse_sse(line)
        print(event['type'])

3. Handle Rate Limits

import time

def submit_with_retry(query, max_retries=3):
    for attempt in range(max_retries):
        response = requests.post(url, json={"query": query})

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            time.sleep(retry_after)
            continue

        return response

4. Use Idempotency Keys

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"query": "Process once"}'

SDKs

Python SDK

Official Python client with streaming support

REST Clients

Use any HTTP client library

Next Steps

Submit Tasks

Task submission API

Stream Events

Real-time event streaming

Sessions API

Session management

Authentication

API keys and JWT authentication

Task Control

Pause, resume, and cancel tasks

Python SDK

Get started with Python

​Introduction

​Quick Start

​API Endpoints

​Core Task Operations

​Task Control

​Real-Time Streaming

​Session Management

​Schedules (Recurring Tasks)

​Authentication

​Approvals

​OpenAI-Compatible API

​Health & Observability

​Authentication

​Response Format

​Rate Limiting

​Event Streaming

​Client Access

​Best Practices

​1. Use Session IDs for Context

​2. Stream Events for Long Tasks

​3. Handle Rate Limits

​4. Use Idempotency Keys

​SDKs

Python SDK

REST Clients

​Next Steps

Submit Tasks

Stream Events

Sessions API

Authentication

Task Control

Python SDK

Introduction

Quick Start

API Endpoints

Core Task Operations

Task Control

Real-Time Streaming

Session Management

Schedules (Recurring Tasks)

Authentication

Approvals

OpenAI-Compatible API

Health & Observability

Authentication

Response Format

Rate Limiting

Event Streaming

Client Access

Best Practices

1. Use Session IDs for Context

2. Stream Events for Long Tasks

3. Handle Rate Limits

4. Use Idempotency Keys

SDKs

Next Steps