Skip to main content

Install

pip install shannon-sdk

First Task

from shannon import ShannonClient
client = ShannonClient(base_url="http://localhost:8080")

handle = client.submit_task("Return the capital of France as a single word.")
final = client.wait(handle.task_id, timeout=60)
print(final.result)

Streaming (SSE)

from shannon import ShannonClient, EventType
client = ShannonClient()

h = client.submit_task(
  "Draft a ~120-word executive summary of three recent quantum computing milestones. Output as a Markdown paragraph."
)
for e in client.stream(h.workflow_id, types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]):
  print(e.type, e.message)
  if e.type == EventType.WORKFLOW_COMPLETED:
    break

Model/Mode Selection (optional)

Control cost and provider choice when submitting:
handle = client.submit_task(
  "Summarize the paragraph into three bullet points focusing on revenue trends. Output: Markdown list.",
  model_tier="small",                # small|medium|large
  # model_override="gpt-5-nano-2025-08-07",
  # provider_override="openai",      # or anthropic, google, ...
  mode="simple",                      # simple|standard|complex|supervisor
)
print(client.wait(handle.task_id).result)

Async Version

import asyncio
from shannon import AsyncShannonClient

async def main():
  async with AsyncShannonClient(base_url="http://localhost:8080") as client:
    h = await client.submit_task(
      "Classify sentiment (positive|neutral|negative) for: 'I love this product!'. Return JSON {label, score}."
    )
    f = await client.wait(h.task_id)
    print(f.result)

asyncio.run(main())

Task Control (Pause/Resume)

Control long-running tasks with pause, resume, and state checking:
from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# Submit a long-running research task
handle = client.submit_task("Research AI trends in 2025 with detailed analysis")

# Pause the task at the next checkpoint
client.pause_task(handle.task_id, reason="Review intermediate results")

# Check control state
state = client.get_control_state(handle.task_id)
print(f"Is paused: {state.is_paused}")
print(f"Paused at: {state.paused_at}")
print(f"Pause reason: {state.pause_reason}")

# Resume execution
client.resume_task(handle.task_id, reason="Continue after review")

# Wait for completion
result = client.wait(handle.task_id)
print(result.result)
CLI Usage:
# Pause a running task
shannon pause task-123 --reason "Hold for review"

# Check control state
shannon control-state task-123

# Resume the task
shannon resume task-123 --reason "Continue"

CLI

# Submit with model selection
python -m shannon.cli --base-url http://localhost:8080 \
  submit "Summarize" --model-tier small --mode simple --wait