Session (primary thread)
├── Coordinator Agent (例: Engineering Lead)
│   ├── → Thread A: Code Reviewer Agent
│   ├── → Thread B: Test Writer Agent
│   └── → Thread C: Security Agent  （必要なら）
│       (同一コンテナ、共有ファイルシステム)
└── Primary thread に全活動の集約ビューが流れる

anthropic>=0.95.0
python-dotenv>=1.0.0
httpx>=0.27.0

import json
import logging
import os
import sys
from datetime import datetime
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

STATE_FILE = Path(__file__).parent / "state.json"
LOGS_DIR = Path(__file__).parent / "logs"

def get_client():
    import anthropic
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        raise ValueError("ANTHROPIC_API_KEY が設定されていません。.env を確認してください。")
    return anthropic.Anthropic(api_key=api_key)

def load_state() -> dict:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}

def save_state(updates: dict) -> dict:
    state = load_state()
    state.update(updates)
    STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))
    return state

def setup_logger(step_name: str) -> logging.Logger:
    LOGS_DIR.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    log_file = LOGS_DIR / f"{step_name}_{timestamp}.txt"

    logger = logging.getLogger(step_name)
    logger.setLevel(logging.DEBUG)
    logger.handlers.clear()

    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S")

    sh = logging.StreamHandler(sys.stdout)
    sh.setFormatter(fmt)
    logger.addHandler(sh)

    fh = logging.FileHandler(log_file, encoding="utf-8")
    fh.setFormatter(fmt)
    logger.addHandler(fh)

    logger.info(f"ログ出力先: {log_file}")
    return logger

from common import get_client, save_state, setup_logger

log = setup_logger("step1")

def main():
    client = get_client()

    log.info("=== Code Reviewer を作成 ===")
    reviewer = client.beta.agents.create(
        name="Code Reviewer",
        model="claude-sonnet-4-6",
        system="""You are a senior code reviewer.
Review code for:
- Security vulnerabilities
- Performance issues
- Code quality and readability
- Best practices

Provide specific, actionable feedback with line numbers.""",
        tools=[{"type": "agent_toolset_20260401"}],
    )
    log.info(f"Reviewer Agent ID: {reviewer.id}")

    log.info("=== Test Writer を作成 ===")
    test_writer = client.beta.agents.create(
        name="Test Writer",
        model="claude-sonnet-4-6",
        system="""You are a test engineer specializing in writing comprehensive unit tests.
Write pytest tests that cover:
- Happy path
- Edge cases
- Error conditions
- Boundary values""",
        tools=[{"type": "agent_toolset_20260401"}],
    )
    log.info(f"Test Writer Agent ID: {test_writer.id}")

    save_state({
        "reviewer_id": reviewer.id,
        "test_writer_id": test_writer.id,
    })
    log.info("state.json に保存しました。次は Step 2 を実行してください。")

if __name__ == "__main__":
    main()

from common import get_client, load_state, save_state, setup_logger

log = setup_logger("step2")

def main():
    state = load_state()
    reviewer_id = state.get("reviewer_id")
    test_writer_id = state.get("test_writer_id")

    if not reviewer_id or not test_writer_id:
        log.error("reviewer_id / test_writer_id がありません。Step 1 を先に実行してください。")
        return

    client = get_client()

    log.info("=== Engineering Lead（コーディネーター）を作成 ===")
    coordinator = client.beta.agents.create(
        name="Engineering Lead",
        model="claude-opus-4-7",
        system="""You are an engineering lead.
When given code to review:
1. First, delegate to the Code Reviewer agent for security and quality review
2. Then, delegate to the Test Writer agent to create tests
3. Synthesize both results into a final report

Always delegate to both agents in parallel when possible.""",
        tools=[{"type": "agent_toolset_20260401"}],
        extra_body={
            "multiagent": {
                "type": "coordinator",
                "agents": [
                    {"type": "agent", "id": reviewer_id},
                    {"type": "agent", "id": test_writer_id},
                ],
            }
        },
    )
    log.info(f"Coordinator Agent ID: {coordinator.id}")

    save_state({"coordinator_id": coordinator.id})
    log.info("state.json を更新しました。次は Step 3 を実行してください。")

if __name__ == "__main__":
    main()

from common import get_client, load_state, save_state, setup_logger

log = setup_logger("step3")

def main():
    state = load_state()
    coordinator_id = state.get("coordinator_id")

    if not coordinator_id:
        log.error("coordinator_id がありません。Step 2 を先に実行してください。")
        return

    client = get_client()

    log.info("=== Environment を作成 ===")
    environment = client.beta.environments.create(name="code-review-env")
    log.info(f"Environment ID: {environment.id}")

    log.info("=== Session を作成 ===")
    session = client.beta.sessions.create(
        agent=coordinator_id,
        environment_id=environment.id,
    )
    log.info(f"Session ID: {session.id}")

    save_state({
        "environment_id": environment.id,
        "session_id": session.id,
    })
    log.info("state.json を更新しました。次は Step 4 を実行してください。")

if __name__ == "__main__":
    main()

from common import get_client, load_state, setup_logger

log = setup_logger("step4")

CODE_TO_REVIEW = '''
def calculate_discount(price, discount_percent):
    discount = price * discount_percent / 100
    final_price = price - discount
    return final_price

def process_order(items, user_id):
    total = 0
    for item in items:
        total += item['price']
    discount = calculate_discount(total, 10)
    return {'user': user_id, 'total': discount}
'''

def main():
    state = load_state()
    session_id = state.get("session_id")

    if not session_id:
        log.error("session_id がありません。Step 3 を先に実行してください。")
        return

    client = get_client()

    log.info("=== タスクを送信 ===")
    client.beta.sessions.events.send(
        session_id,
        events=[{
            "type": "user.message",
            "content": [{
                "type": "text",
                "text": (
                    "Please review this Python code and generate tests:\n\n"
                    f"```python\n{CODE_TO_REVIEW}\n```"
                ),
            }],
        }],
    )
    log.info("送信済み。次は Step 5 でストリームを観察してください。")

if __name__ == "__main__":
    main()

from common import get_client, load_state, setup_logger

log = setup_logger("step5")

def main():
    state = load_state()
    session_id = state.get("session_id")

    if not session_id:
        log.error("session_id がありません。Step 3 を先に実行してください。")
        return

    log.info(f"Session ID: {session_id}")
    log.info("=== Primary Thread のイベントストリーム ===")

    client = get_client()

    with client.beta.sessions.events.stream(session_id) as stream:
        for event in stream:
            match event.type:
                case "session.thread_created":
                    agent_name = getattr(event, "agent_name", "unknown")
                    log.info(f">> 新スレッド作成: {agent_name}")

                case "session.thread_status_running":
                    agent_name = getattr(event, "agent_name", "")
                    log.info(f">> スレッド実行中: {agent_name}")

                case "session.thread_status_idle":
                    agent_name = getattr(event, "agent_name", "")
                    session_thread_id = getattr(event, "session_thread_id", None)
                    stop_reason = getattr(event, "stop_reason", None)
                    log.info(f">> スレッド待機: {agent_name} | stop_reason={stop_reason}")
                    if session_thread_id is None:
                        log.info(">> Primary が idle → セッション完了")
                        break

                case "agent.thread_message_received":
                    from_agent = getattr(event, "from_agent_name", "unknown")
                    content = getattr(event, "content", [])
                    log.info(f"\n>> {from_agent} から結果を受信:")
                    for block in content:
                        if getattr(block, "type", None) == "text":
                            preview = block.text[:400]
                            log.info(f"   {preview}{'...' if len(block.text) > 400 else ''}")
                            break

                case "agent.thread_message_sent":
                    to_agent = getattr(event, "to_agent_name", "unknown")
                    log.info(f">> コーディネーターが {to_agent} に委任")

                case _:
                    log.debug(f"   [event] {event.type}")

    log.info("=== ストリーム終了 ===")

if __name__ == "__main__":
    main()

import os

import httpx
from dotenv import load_dotenv

from common import load_state, setup_logger

log = setup_logger("step6")

BASE_URL = "https://api.anthropic.com/v1"
BETA_HEADER = "managed-agents-2026-04-01"

def get_headers() -> dict:
    return {
        "x-api-key": os.environ["ANTHROPIC_API_KEY"],
        "anthropic-version": "2023-06-01",
        "anthropic-beta": BETA_HEADER,
        "content-type": "application/json",
    }

def list_threads(session_id: str) -> list:
    resp = httpx.get(
        f"{BASE_URL}/sessions/{session_id}/threads",
        headers=get_headers(),
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])

def list_thread_events(session_id: str, thread_id: str) -> list:
    resp = httpx.get(
        f"{BASE_URL}/sessions/{session_id}/threads/{thread_id}/events",
        headers=get_headers(),
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])

def main():
    load_dotenv()

    if not os.environ.get("ANTHROPIC_API_KEY"):
        log.error("ANTHROPIC_API_KEY が設定されていません。")
        return

    state = load_state()
    session_id = state.get("session_id")

    if not session_id:
        log.error("session_id がありません。Step 3 を先に実行してください。")
        return

    log.info("=== 全スレッド一覧 ===")
    threads = list_threads(session_id)
    log.info(f"合計スレッド数: {len(threads)}")

    for thread in threads:
        agent_name = thread.get("agent", {}).get("name", "unknown")
        status = thread.get("status", "unknown")
        parent = thread.get("parent_thread_id", None)
        thread_id = thread.get("id", "")
        log.info(f"  [{agent_name}] id={thread_id} status={status} | parent={parent}")

    log.info("")
    log.info("=== 各スレッドのイベント ===")
    for thread in threads:
        agent_name = thread.get("agent", {}).get("name", "unknown")
        thread_id = thread.get("id", "")
        log.info(f"\n--- {agent_name} (id={thread_id}) ---")

        events = list_thread_events(session_id, thread_id)
        if not events:
            log.info("  (イベントなし)")
            continue

        for ev in events:
            log.info(f"  [{ev.get('type', 'unknown')}] {ev.get('processed_at', '')}")

    log.info("\n=== 完了 ===")

if __name__ == "__main__":
    main()

05:57:25 [DEBUG]  [event] agent.thinking
05:57:27 [DEBUG]  [event] span.model_request_end
05:57:27 [INFO]  >> 新スレッド作成: Code Reviewer
05:57:27 [INFO]  >> コーディネーターが Code Reviewer に委任
05:57:27 [INFO]  >> 新スレッド作成: Test Writer
05:57:27 [INFO]  >> コーディネーターが Test Writer に委任
05:57:27 [INFO]  >> スレッド実行中: Code Reviewer
05:57:27 [INFO]  >> スレッド実行中: Test Writer
05:58:12 [INFO]  >> Code Reviewer から結果を受信:
                    ## Code Review Report
                    ...（コード品質・セキュリティ指摘）
05:59:33 [INFO]  >> Test Writer から結果を受信:
                    Here is the complete `test_orders.py` (45 tests, all passing...)
06:00:44 [INFO]  >> スレッド待機: Engineering Lead | stop_reason={'type': 'end_turn'}
06:00:44 [DEBUG]  [event] session.status_idle

[Engineering Lead] id=sthr_01xxxx... status=idle | parent=None
[Code Reviewer]    id=sthr_01yyyy... status=idle | parent=sthr_01xxxx...
[Test Writer]      id=sthr_01zzzz... status=idle | parent=sthr_01xxxx...