クイックスタート

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)

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)

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)

# 実行中のタスクを一時停止
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']}")

# レビュー状態を取得
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}")

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

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

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

# モデル選択で提出
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