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

​
Tool API

List, inspect, and execute tools directly: 
from shannon import ShannonClient

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

# List available tools
tools = client.list_tools()
for tool in tools:
    print(f"{tool.name}: {tool.description}")

# Get tool details
detail = client.get_tool("web_search")
print(detail.parameters)

# Execute a tool directly
result = client.execute_tool("web_search", arguments={"query": "Shannon AI"})
print(result.output)
CLI Usage: 
# List tools
shannon tools-list

# Get tool details
shannon tool-get web_search

# Execute a tool
shannon tool-exec web_search --arguments '{"query": "Shannon AI"}'

​
Agents API

List, inspect, and execute deterministic agents: 
from shannon import ShannonClient

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

# List deterministic agents
agents = client.list_agents()

# Get agent details
agent = client.get_agent(agents[0].id)
print(f"{agent.name}: {agent.description}")

# Execute an agent
execution = client.execute_agent(agent.id, {"query": "test input"})
status = execution.wait(timeout=60)
print(status.result)

# Send follow-up message to a running swarm
result = client.send_swarm_message(workflow_id, "Please focus on cost analysis")
CLI Usage: 
# List agents
shannon agents-list

# Get agent details
shannon agent-get AGENT_ID

# Execute an agent
shannon agent-exec AGENT_ID --input '{"query": "test input"}'

# Send follow-up to swarm
shannon swarm-message WORKFLOW_ID "Please focus on cost analysis"

​
OpenAI-Compatible

Use the OpenAI-compatible chat completions endpoint: 
from shannon import ShannonClient

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

# List available models
models = client.list_openai_models()

# Chat completion (non-streaming)
completion = client.create_chat_completion(
    messages=[{"role": "user", "content": "Hello!"}],
    model="gpt-4o-mini",
)
print(completion.choices[0].message.content)

# Streaming chat completion
for chunk in client.stream_chat_completion(
    messages=[{"role": "user", "content": "Tell me a story"}]
):
    if chunk.choices and chunk.choices[0].delta:
        print(chunk.choices[0].delta.content, end="")

​
File Access

Access session workspace files and user memory files: 
from shannon import ShannonClient

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

# List workspace files from a session
files = client.list_session_files(session_id)
for f in files:
    print(f"{f.name} ({f.size_bytes} bytes)")

# Download a file
content = client.download_session_file(session_id, "report.md")
print(content.content)

# List and download memory files
memory_files = client.list_memory_files()
memory = client.download_memory_file("profile.md")
CLI Usage: 
# List session workspace files
shannon session-files SESSION_ID

# Download a workspace file
shannon session-file-get SESSION_ID report.md

# List memory files
shannon memory-files

# Download a memory file
shannon memory-file-get profile.md

​
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