Quick Start Examples

Get started with Shannon API in minutes - complete examples for your first API calls

Your First API Call

The basic Shannon workflow is simple: submit a task, get a task ID, check the status, and retrieve the result.

Complete Example (cURL)

# 1. Submit a task
RESPONSE=$(curl -sS -X POST http://localhost:8080/api/v1/tasks \
  -H "X-API-Key: sk_test_123456" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "What is the capital of France?"
  }')

# 2. Extract task ID
TASK_ID=$(echo $RESPONSE | jq -r '.task_id')
echo "Task ID: $TASK_ID"

# 3. Poll for result
while true; do
  STATUS=$(curl -sS http://localhost:8080/api/v1/tasks/$TASK_ID \
    -H "X-API-Key: sk_test_123456")

  STATE=$(echo $STATUS | jq -r '.status')

  if [ "$STATE" = "TASK_STATUS_COMPLETED" ]; then
    echo $STATUS | jq -r '.result'
    break
  elif [ "$STATE" = "TASK_STATUS_FAILED" ]; then
    echo "Error: $(echo $STATUS | jq -r '.error')"
    exit 1
  fi

  sleep 2
done

Expected Output:

Task ID: task_01HQZX3Y9K8M2P4N5S7T9W2V
The capital of France is Paris.

Common API Patterns

1. Submit → Poll → Retrieve

The standard workflow for any task:

Python

import httpx
import time

# Submit task
response = httpx.post(
    "http://localhost:8080/api/v1/tasks",
    headers={"X-API-Key": "sk_test_123456"},
    json={"query": "What is the capital of France?"}
)
task_id = response.json()["task_id"]
print(f"Task submitted: {task_id}")

# Poll until complete
while True:
    status_resp = httpx.get(
        f"http://localhost:8080/api/v1/tasks/{task_id}",
        headers={"X-API-Key": "sk_test_123456"}
    )
    status = status_resp.json()

    if status["status"] == "TASK_STATUS_COMPLETED":
        print("Result:", status["result"])
        break
    elif status["status"] == "TASK_STATUS_FAILED":
        print("Error:", status["error"])
        break

    time.sleep(2)

2. Multi-Turn Conversations (Sessions)

Use the same session_id to maintain context across multiple tasks:

Python

import httpx

API_KEY = "sk_test_123456"
SESSION_ID = "user-123-chat"

# Turn 1
resp1 = httpx.post(
    "http://localhost:8080/api/v1/tasks",
    headers={"X-API-Key": API_KEY},
    json={
        "query": "What is Python?",
        "session_id": SESSION_ID
    }
)
print("Task 1:", resp1.json()["task_id"])

# Turn 2 (references previous context)
resp2 = httpx.post(
    "http://localhost:8080/api/v1/tasks",
    headers={"X-API-Key": API_KEY},
    json={
        "query": "What are its main advantages?",
        "session_id": SESSION_ID  # Same session
    }
)
print("Task 2:", resp2.json()["task_id"])

3. Real-Time Streaming (SSE)

Get live updates instead of polling:

Python

import httpx

# Submit task
response = httpx.post(
    "http://localhost:8080/api/v1/tasks",
    headers={"X-API-Key": "sk_test_123456"},
    json={"query": "Write a haiku about coding"}
)
task_id = response.json()["task_id"]

# Stream events
with httpx.stream(
    "GET",
    f"http://localhost:8080/api/v1/tasks/{task_id}/stream",
    headers={"X-API-Key": "sk_test_123456"}
) as stream:
    for line in stream.iter_lines():
        if line.startswith("data: "):
            event_data = line[6:]  # Remove "data: " prefix
            print(event_data)

4. Error Handling

Always handle errors gracefully:

Python

import httpx

try:
    response = httpx.post(
        "http://localhost:8080/api/v1/tasks",
        headers={"X-API-Key": "sk_test_123456"},
        json={"query": "Analyze this data"},
        timeout=30.0
    )
    response.raise_for_status()

    task = response.json()
    print(f"Success: {task['task_id']}")

except httpx.HTTPStatusError as e:
    if e.response.status_code == 401:
        print("Error: Invalid API key")
    elif e.response.status_code == 429:
        retry_after = e.response.headers.get("Retry-After", "60")
        print(f"Rate limited. Retry after {retry_after}s")
    else:
        print(f"HTTP error: {e.response.status_code}")
        print(e.response.json())

except httpx.TimeoutException:
    print("Request timed out")

except Exception as e:
    print(f"Unexpected error: {e}")

Authentication Setup

Getting Your API Key

1. Development Mode (Skip Auth)

   For local testing, you can disable authentication:

   # In .env file
   GATEWAY_SKIP_AUTH=1
   

   Then use any placeholder key:

   curl -H "X-API-Key: dev" http://localhost:8080/api/v1/tasks
   
2. Production Mode (Real API Keys)

   Generate an API key from your Shannon dashboard:

   1. Log in to Shannon Cloud
   2. Navigate to Settings → API Keys
   3. Click “Create API Key”
   4. Copy the key (starts with sk_)

   Store it securely (e.g., environment variable):

   export SHANNON_API_KEY="sk_live_abc123..."
   

Including the API Key

Always include your API key in the X-API-Key header:

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

# ❌ Wrong - Authorization header doesn't work for API keys
curl -H "Authorization: Bearer sk_test_123456" http://localhost:8080/api/v1/tasks

Never commit API keys to version control. Use environment variables or secret management tools.

Complete Workflow Example

Here’s a real-world example that combines all patterns:

Python

import httpx
import time
import os

class ShannonClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key

    def submit_task(self, query: str, session_id: str = None):
        """Submit a task and return task ID."""
        response = httpx.post(
            f"{self.base_url}/api/v1/tasks",
            headers={"X-API-Key": self.api_key},
            json={
                "query": query,
                "session_id": session_id
            }
        )
        response.raise_for_status()
        return response.json()["task_id"]

    def wait_for_completion(self, task_id: str, timeout: int = 300):
        """Poll until task completes."""
        start_time = time.time()

        while True:
            if time.time() - start_time > timeout:
                raise TimeoutError(f"Task timeout after {timeout}s")

            response = httpx.get(
                f"{self.base_url}/api/v1/tasks/{task_id}",
                headers={"X-API-Key": self.api_key}
            )
            status = response.json()

            if status["status"] == "TASK_STATUS_COMPLETED":
                return status["result"]
            elif status["status"] == "TASK_STATUS_FAILED":
                raise Exception(f"Task failed: {status['error']}")

            time.sleep(2)

    def run_task(self, query: str, session_id: str = None):
        """Submit and wait for result in one call."""
        task_id = self.submit_task(query, session_id)
        return self.wait_for_completion(task_id)

# Usage
client = ShannonClient(
    base_url="http://localhost:8080",
    api_key=os.getenv("SHANNON_API_KEY", "sk_test_123456")
)

# Single task
result = client.run_task("What is the capital of France?")
print(result)

# Multi-turn conversation
session_id = "chat-123"
result1 = client.run_task("What is Python?", session_id)
result2 = client.run_task("What are its advantages?", session_id)
print(result1)
print(result2)

Next Steps

Now that you know the basics, explore more advanced features:

Full API Reference

Complete documentation for all endpoints

Agents Catalog

Pre-built agents for specific tasks

Python SDK

Official Python client library

Streaming API

Real-time event streaming with SSE

Common Questions

How do I know when my task is complete?

Poll the GET /api/v1/tasks/{id} endpoint until status is TASK_STATUS_COMPLETED or TASK_STATUS_FAILED. For long-running tasks, use streaming instead of polling.

What's the difference between result and response?

• result is the raw text output from the LLM
• response is only present if result contains valid JSON (parsed automatically)

Most of the time, you’ll use result.

How long do sessions last?

Sessions persist for 30 days by default. All tasks with the same session_id share conversation history.

Can I cancel a running task?

Yes, use POST /api/v1/tasks/{id}/cancel. The task will stop at the next safe checkpoint.

Why use streaming instead of polling?

Streaming provides real-time updates with lower latency and less server load. Use it for long-running tasks or when you need immediate feedback.

Troubleshooting

401 Unauthorized

• Check your API key is correct
• Verify you’re using the X-API-Key header (not Authorization)
• In dev mode, ensure GATEWAY_SKIP_AUTH=1 is set

429 Rate Limited

• Check the Retry-After header
• Implement exponential backoff
• Upgrade your API tier for higher limits

Task stuck in RUNNING

• Wait longer (complex tasks take time)
• Use streaming to see progress
• Check Temporal UI for detailed workflow status

Empty result field

• Task might still be running (check status)
• Task failed (check error field)
• Use streaming to see intermediate output