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)

​
Swarm Mode

Force multi-agent swarm execution with force_swarm: 
handle = client.submit_task(
  "Research and compare the top 3 cloud providers for AI workloads.",
  force_swarm=True,
)
result = client.wait(handle.task_id)
print(result.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"

​
Human-in-the-Loop Review

Interact with workflows that require human review before proceeding: 
from shannon import ShannonClient

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

# Get the current review state for a workflow
state = client.get_review_state("workflow-123")
print(f"Status: {state.status}")        # "reviewing" or "approved"
print(f"Round: {state.round}")
print(f"Current plan: {state.current_plan}")

# View conversation rounds
for r in state.rounds:
    print(f"  [{r.role}] {r.message}")

# Submit feedback to refine the plan
updated = client.submit_review_feedback(
    "workflow-123",
    "Please add error handling for network failures.",
    version=state.version,  # optimistic concurrency
)
print(f"Updated round: {updated.round}")

# Approve the plan to let the workflow proceed
result = client.approve_review("workflow-123", version=updated.version)
print(f"Review status: {result['status']}")
CLI Usage: 
# Get review state
shannon review-get workflow-123

# Submit feedback
shannon review-feedback workflow-123 "Add error handling"

# Approve
shannon review-approve workflow-123

​
Skills

List and inspect available skills: 
from shannon import ShannonClient

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

# List all skills
skills = client.list_skills()
for s in skills:
    print(f"{s.name} v{s.version} [{s.category}] - {s.description}")

# Filter by category
coding_skills = client.list_skills(category="coding")

# Get detailed information about a specific skill
detail = client.get_skill("web_search")
print(f"Author: {detail.author}")
print(f"Requires tools: {detail.requires_tools}")
print(f"Content: {detail.content}")

# Get all versions of a skill
versions = client.get_skill_versions("web_search")
for v in versions:
    print(f"  {v.name} v{v.version}")
CLI Usage: 
# List skills
shannon skills-list

# Get skill details
shannon skill-get web_search

# Get skill versions
shannon skill-versions web_search

​
CLI

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

# Submit with swarm mode
python -m shannon.cli --base-url http://localhost:8080 \
  submit "Research cloud providers" --swarm --wait