クイックスタート

インストール

pip install shannon-sdk

最初のタスク

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

handle = client.submit_task("フランスの首都を一言で返してください。")
final = client.wait(handle.task_id, timeout=60)
print(final.result)

ストリーミング (SSE)

from shannon import ShannonClient, EventType
client = ShannonClient()

h = client.submit_task(
  "最近の量子コンピューティングのマイルストーン3つについて、約120語のエグゼクティブサマリーを作成してください。Markdownの段落として出力します。"
)
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

モデル/モード選択 (オプション)

提出時にコストとプロバイダーの選択を制御します：

handle = client.submit_task(
  "段落を収益トレンドに焦点を当てて3つの箇条書きに要約してください。出力: Markdownリスト。",
  model_tier="small",                # small|medium|large
  # model_override="gpt-5-nano-2025-08-07",
  # provider_override="openai",      # または anthropic, google, ...
  mode="simple",                      # simple|standard|complex|supervisor
)
print(client.wait(handle.task_id).result)

スウォームモード

force_swarm でマルチエージェントスウォーム実行を強制します：

handle = client.submit_task(
  "AIワークロード向けの上位3つのクラウドプロバイダーを調査・比較してください。",
  force_swarm=True,
)
result = client.wait(handle.task_id)
print(result.result)

非同期バージョン

import asyncio
from shannon import AsyncShannonClient

async def main():
  async with AsyncShannonClient(base_url="http://localhost:8080") as client:
    h = await client.submit_task(
      "次の文の感情を分類してください (positive|neutral|negative): 'この製品が大好きです！'。JSON {label, score}を返してください。"
    )
    f = await client.wait(h.task_id)
    print(f.result)

asyncio.run(main())

タスク制御 (一時停止/再開)

長時間実行されるタスクを一時停止、再開、状態確認で制御します：

from shannon import ShannonClient

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

# 長時間実行される研究タスクを提出
handle = client.submit_task("2025年のAIトレンドについて詳細な分析を行う")

# 次のチェックポイントでタスクを一時停止
client.pause_task(handle.task_id, reason="中間結果のレビュー")

# 制御状態を確認
state = client.get_control_state(handle.task_id)
print(f"一時停止中: {state.is_paused}")
print(f"一時停止時: {state.paused_at}")
print(f"一時停止理由: {state.pause_reason}")

# 実行を再開
client.resume_task(handle.task_id, reason="レビュー後に続行")

# 完了を待つ
result = client.wait(handle.task_id)
print(result.result)

CLIの使用法:

# 実行中のタスクを一時停止
shannon pause task-123 --reason "レビューのため保留"

# 制御状態を確認
shannon control-state task-123

# タスクを再開
shannon resume task-123 --reason "続行"

ヒューマン・イン・ザ・ループレビュー

人間のレビューが必要なワークフローと対話します：

from shannon import ShannonClient

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

# ワークフローの現在のレビュー状態を取得
state = client.get_review_state("workflow-123")
print(f"ステータス: {state.status}")        # "reviewing" または "approved"
print(f"ラウンド: {state.round}")
print(f"現在のプラン: {state.current_plan}")

# 会話ラウンドを表示
for r in state.rounds:
    print(f"  [{r.role}] {r.message}")

# フィードバックを送信してプランを改善
updated = client.submit_review_feedback(
    "workflow-123",
    "ネットワーク障害のエラーハンドリングを追加してください。",
    version=state.version,  # 楽観的同時実行制御
)
print(f"更新されたラウンド: {updated.round}")

# プランを承認してワークフローを続行
result = client.approve_review("workflow-123", version=updated.version)
print(f"レビューステータス: {result['status']}")

CLIの使用法:

# レビュー状態を取得
shannon review-get workflow-123

# フィードバックを送信
shannon review-feedback workflow-123 "エラーハンドリングを追加"

# 承認
shannon review-approve workflow-123

スキル

利用可能なスキルの一覧と詳細の確認：

from shannon import ShannonClient

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

# すべてのスキルを一覧表示
skills = client.list_skills()
for s in skills:
    print(f"{s.name} v{s.version} [{s.category}] - {s.description}")

# カテゴリでフィルタリング
coding_skills = client.list_skills(category="coding")

# 特定のスキルの詳細情報を取得
detail = client.get_skill("web_search")
print(f"著者: {detail.author}")
print(f"必要なツール: {detail.requires_tools}")
print(f"コンテンツ: {detail.content}")

# スキルのすべてのバージョンを取得
versions = client.get_skill_versions("web_search")
for v in versions:
    print(f"  {v.name} v{v.version}")

CLIの使用法:

# スキルを一覧表示
shannon skills-list

# スキルの詳細を取得
shannon skill-get web_search

# スキルのバージョンを取得
shannon skill-versions web_search

ツール API

ツールの一覧表示、詳細確認、直接実行：

from shannon import ShannonClient

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

# 利用可能なツールを一覧表示
tools = client.list_tools()
for tool in tools:
    print(f"{tool.name}: {tool.description}")

# ツールの詳細を取得
detail = client.get_tool("web_search")
print(detail.parameters)

# ツールを直接実行
result = client.execute_tool("web_search", arguments={"query": "Shannon AI"})
print(result.output)

CLIの使用法:

# ツールを一覧表示
shannon tools-list

# ツールの詳細を取得
shannon tool-get web_search

# ツールを実行
shannon tool-exec web_search --arguments '{"query": "Shannon AI"}'

エージェント API

確定的エージェントの一覧表示、詳細確認、実行：

from shannon import ShannonClient

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

# 確定的エージェントを一覧表示
agents = client.list_agents()

# エージェントの詳細を取得
agent = client.get_agent(agents[0].id)
print(f"{agent.name}: {agent.description}")

# エージェントを実行
execution = client.execute_agent(agent.id, {"query": "test input"})
status = execution.wait(timeout=60)
print(status.result)

# 実行中のスウォームにフォローアップメッセージを送信
result = client.send_swarm_message(workflow_id, "Please focus on cost analysis")

CLIの使用法:

# エージェントを一覧表示
shannon agents-list

# エージェントの詳細を取得
shannon agent-get AGENT_ID

# エージェントを実行
shannon agent-exec AGENT_ID --input '{"query": "test input"}'

# スウォームにフォローアップを送信
shannon swarm-message WORKFLOW_ID "Please focus on cost analysis"

OpenAI互換

OpenAI互換のチャット補完エンドポイントを使用：

from shannon import ShannonClient

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

# 利用可能なモデルを一覧表示
models = client.list_openai_models()

# チャット補完（非ストリーミング）
completion = client.create_chat_completion(
    messages=[{"role": "user", "content": "Hello!"}],
    model="gpt-4o-mini",
)
print(completion.choices[0].message.content)

# ストリーミングチャット補完
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="")

ファイルアクセス

セッションワークスペースファイルとユーザーメモリファイルにアクセス：

from shannon import ShannonClient

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

# セッションのワークスペースファイルを一覧表示
files = client.list_session_files(session_id)
for f in files:
    print(f"{f.name} ({f.size_bytes} bytes)")

# ファイルをダウンロード
content = client.download_session_file(session_id, "report.md")
print(content.content)

# メモリファイルの一覧表示とダウンロード
memory_files = client.list_memory_files()
memory = client.download_memory_file("profile.md")

CLIの使用法:

# セッションワークスペースファイルを一覧表示
shannon session-files SESSION_ID

# ワークスペースファイルをダウンロード
shannon session-file-get SESSION_ID report.md

# メモリファイルを一覧表示
shannon memory-files

# メモリファイルをダウンロード
shannon memory-file-get profile.md

CLI

# モデル選択で提出
python -m shannon.cli --base-url http://localhost:8080 \
  submit "要約" --model-tier small --mode simple --wait

# スウォームモードで提出
python -m shannon.cli --base-url http://localhost:8080 \
  submit "クラウドプロバイダーを調査" --swarm --wait

Python SDK

​インストール

​最初のタスク

​ストリーミング (SSE)

​モデル/モード選択 (オプション)

​スウォームモード

​非同期バージョン

​タスク制御 (一時停止/再開)

​ヒューマン・イン・ザ・ループ レビュー

​スキル

​ツール API

​エージェント API

​OpenAI互換

​ファイルアクセス

​CLI

インストール

最初のタスク

ストリーミング (SSE)

モデル/モード選択 (オプション)

スウォームモード

非同期バージョン

タスク制御 (一時停止/再開)

ヒューマン・イン・ザ・ループレビュー

スキル

ツール API

エージェント API

OpenAI互換

ファイルアクセス

CLI