Claude Managed Agents API実践ガイド — セルフホストCLI運用者が語る「どこまで移行すべきか」の判断フロー

from anthropic import Anthropic

client = Anthropic()

# 1. Agent定義
agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-opus-4-7",
    system="You are a helpful coding assistant.",
    tools=[{"type": "agent_toolset_20260401"}],
)

# 2. Environment作成
environment = client.beta.environments.create(
    name="dev-env",
    config={"type": "cloud", "networking": {"type": "unrestricted"}},
)

# 3. Session開始
session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="My first session",
)

agent = client.beta.agents.create(
    name="Restricted Agent",
    model="claude-opus-4-7",
    tools=[{
        "type": "agent_toolset_20260401",
        "configs": [
            {"name": "web_fetch", "enabled": False},
            {"name": "web_search", "enabled": False},
        ],
    }],
)

{
  "type": "agent_toolset_20260401",
  "default_config": { "enabled": false },
  "configs": [
    { "name": "bash", "enabled": true },
    { "name": "read", "enabled": true },
    { "name": "write", "enabled": true }
  ]
}

agent = client.beta.agents.create(
    name="Deploy Agent",
    model="claude-opus-4-7",
    tools=[
        {"type": "agent_toolset_20260401"},
        {
            "type": "custom",
            "name": "deploy_to_production",
            "description": "Deploy the current build to production. Requires human approval.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "service": {"type": "string", "description": "Service name to deploy"},
                    "version": {"type": "string", "description": "Version tag"},
                },
                "required": ["service", "version"],
            },
        },
    ],
)

vault = client.beta.vaults.create(
    display_name="User credentials",
    metadata={"internal_user_id": "u_001", "team": "engineering"},
)

credential = client.beta.vaults.credentials.create(
    vault_id=vault.id,
    display_name="GitHub Token",
    auth={
        "type": "static_bearer",
        "mcp_server_url": "https://api.githubcopilot.com/mcp/",
        "token": GITHUB_TOKEN,
    },
)

# Session作成時にVaultを紐付け
session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    vault_ids=[vault.id],
)

from fastapi import FastAPI, Header, HTTPException, Request
import hmac, hashlib, json

app = FastAPI()
WEBHOOK_SECRET = "whsec_..."  # シークレットマネージャーから読み込む

def verify_signature(body: bytes, sig: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, sig)

@app.post("/webhooks/anthropic")
async def handle_webhook(req: Request, x_anthropic_signature: str = Header()):
    body = await req.body()
    if not verify_signature(body, x_anthropic_signature):
        raise HTTPException(401)

    event = json.loads(body)
    session_id = event["resource_id"]

    if event["event_type"] == "session.status_idled":
        events = client.beta.sessions.events.list(session_id=session_id)
        pending = [
            e for e in events.data
            if e.type == "agent.custom_tool_use" and e.name == "escalate"
        ]
        if pending:
            for tu in pending:
                # Slack通知 → 人間が承認 → 結果を返す
                enqueue_for_review(session_id, tu.id, tu.input)
    return {"ok": True}

def submit_approval(session_id: str, tool_use_id: str, decision: str):
    client.beta.sessions.events.send(
        session_id=session_id,
        events=[{
            "type": "user.custom_tool_result",
            "custom_tool_use_id": tool_use_id,
            "content": [{"type": "text", "text": json.dumps({"approved": decision})}],
        }],
    )

# メモリストア作成
store = client.beta.memory_stores.create(
    name="Project Context",
    description="過去のセッションで学んだプロジェクト固有の知識",
)

# 初期データの投入（オプション）
client.beta.memory_stores.memories.create(
    store.id,
    path="/conventions.md",
    content="デプロイ先は常にVercel。DBはSupabase。テストはvitest。",
)

# セッション作成時にメモリストアを添付
session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    resources=[{
        "type": "memory_store",
        "memory_store_id": store.id,
        "access": "read_write",
        "instructions": "タスク開始前に必ず確認すること。新しい学びがあれば書き込むこと。",
    }],
)