Managed Agents - Python / 托管智能体 - Python

Bindings not shown here: This README covers the most common managed-agents flows for Python. If you need a class, method, namespace, field, or behavior that isn't shown, WebFetch the Python SDK repo or the relevant docs page from shared/live-sources.md rather than guess. Do not extrapolate from cURL shapes or another language's SDK.

此处未展示的绑定： 本 README 涵盖 Python 上最常见的托管智能体（managed-agents）流程。如果你需要的类、方法、命名空间、字段或行为未在此展示，请通过 WebFetch 访问 shared/live-sources.md 中的 Python SDK 仓库或相关文档页面，而不是凭猜测行事。不要从 cURL 形态或其他语言的 SDK 进行外推。

【评论】该段落是典型的防幻觉指令：要求模型在缺少依据时实时抓取官方源码或文档，而不是基于模式相似性编造 API 形态。

Agents are persistent - create once, reference by ID. Store the agent ID returned by agents.create and pass it to every subsequent sessions.create; do not call agents.create in the request path. Recommended: define agents and environments as version-controlled files synced with ant apply - see shared/anthropic-cli.md (its live-docs URL is in shared/live-sources.md). The CLI owns the control plane (create/update); your code owns the data plane (sessions with the stored ID). The examples below show in-code creation for when you must provision programmatically; in production the create call belongs in setup, not in the request path.

智能体是持久的 —— 创建一次，按 ID 引用。 保存 agents.create 返回的智能体 ID，并在之后每次 sessions.create 时传入它；不要在请求路径中调用 agents.create。推荐做法： 把智能体和环境定义为受版本控制的文件，用 ant apply 同步 —— 见 shared/anthropic-cli.md（其实时文档 URL 在 shared/live-sources.md 中）。CLI 负责控制平面（创建/更新）；你的代码负责数据平面（用已保存的 ID 发起会话）。下面的示例展示了必须在代码中以编程方式创建资源时的写法；在生产环境中，创建调用应属于初始化设置，而不是请求路径。

Installation / 安装

pip install anthropic

Client Initialization / 客户端初始化

import anthropic

# Default - resolves credentials from the environment:
# ANTHROPIC_API_KEY, or ANTHROPIC_AUTH_TOKEN, or an `ant auth login` profile.
# Prefer this for local dev; don't hardcode a key.
client = anthropic.Anthropic()

# Explicit API key (only when you must inject a specific key)
client = anthropic.Anthropic(api_key="your-api-key")

Create an Environment / 创建环境

environment = client.beta.environments.create(
    name="my-dev-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)
print(environment.id)  # env_...

Create an Agent (required first step) / 创建智能体（必需的第一步）

Warning: There is no inline agent config. model/system/tools live on the agent object, not the session. Always start with agents.create() - the session only takes agent={"type": "agent", "id": agent.id}.

警告：不存在内联的智能体配置。 model/system/tools 位于智能体对象上，而不是会话上。始终从 agents.create() 开始 —— 会话只接受 agent={"type": "agent", "id": agent.id}。

Minimal / 最小示例

# 1. Create the agent (reusable, versioned)
agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-opus-5-5",
    tools=[{"type": "agent_toolset_20260401", "default_config": {"enabled": True}}],
)

# 2. Start a session
session = client.beta.sessions.create(
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    environment_id=environment.id,
)
print(session.id, session.status)
print(f"Trace: https://platform.claude.com/workspaces/default/sessions/{session.id}")  # swap 'default' for your workspace ID if the API key is not in the Default workspace

With system prompt and custom tools / 带系统提示词与自定义工具

import os

agent = client.beta.agents.create(
    name="Code Reviewer",
    model="claude-opus-5-5",
    system="You are a senior code reviewer.",
    tools=[
        {"type": "agent_toolset_20260401"},
        {
            "type": "custom",
            "name": "run_tests",
            "description": "Run the test suite",
            "input_schema": {
                "type": "object",
                "properties": {
                    "test_path": {"type": "string", "description": "Path to test file"}
                },
                "required": ["test_path"],
            },
        },
    ],
)

session = client.beta.sessions.create(
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    environment_id=environment.id,
    title="Code review session",
    resources=[
        {
            "type": "github_repository",
            "url": "https://github.com/owner/repo",
            "mount_path": "/workspace/repo",
            "authorization_token": os.environ["GITHUB_TOKEN"],
            "branch": "main",
        }
    ],
)

Send a User Message / 发送用户消息

client.beta.sessions.events.send(
    session_id=session.id,
    events=[
        {
            "type": "user.message",
            "content": [{"type": "text", "text": "Review the auth module"}],
        }
    ],
)

Tip: Stream-first: Open the stream before (or concurrently with) sending the message. The stream only delivers events that occur after it opens - stream-after-send means early events arrive buffered in one batch. See Steering Patterns.

提示：流式优先： 在发送消息之前（或同时）打开流。流只会传递在它打开之后发生的事件 —— 先发送后开流意味着早期事件会以一批缓冲的形式一次性到达。见 Steering Patterns。

Define an Outcome (default kickoff for deliverables) / 定义成果（交付物的默认启动方式）

When the session's job is to produce something checkable - an artifact, a report, a PR - kick off with user.define_outcome instead of user.message: the harness grades each iteration against your rubric and the agent revises until it passes. Send one or the other, never both. See Outcomes for the event reference and rubric-writing guidance.

当会话的任务是产出可检验的东西 —— 一个产物、一份报告、一个 PR —— 时，用 user.define_outcome 而不是 user.message 来启动：执行框架（harness）会按你的评分标准（rubric）对每次迭代打分，智能体不断修订直到通过。二者只发其一，绝不同时发送。事件参考和评分标准撰写指南见 Outcomes。

STARTER_RUBRIC = """# Report rubric - starter, tune the criteria
- Output is a single `report.md` in /mnt/session/outputs/
- Every claim cites a source URL
- Includes a summary table with one row per competitor
- Prices are current as of the run date and each row says where it was read from
- No placeholder text, TODOs, or empty sections remain
"""

client.beta.sessions.events.send(
    session_id=session.id,
    events=[
        {
            "type": "user.define_outcome",
            "description": "Write a competitor-pricing report as report.md",
            "rubric": {"type": "text", "content": STARTER_RUBRIC},
            "max_iterations": 5,  # optional; default 3, max 20
        }
    ],
)

Stream Events (SSE) / 流式接收事件（SSE）

import json

# Stream-first: open stream, then send while stream is live
with client.beta.sessions.events.stream(
    session_id=session.id,
) as stream:
    client.beta.sessions.events.send(
        session_id=session.id,
        events=[{"type": "user.message", "content": [{"type": "text", "text": "..."}]}],
    )
    for event in stream:
        ...  # process events

# Standalone stream iteration:
with client.beta.sessions.events.stream(
    session_id=session.id,
) as stream:
    for event in stream:
        if event.type == "agent.message":
            for block in event.content:
                if block.type == "text":
                    print(block.text, end="", flush=True)
        elif event.type == "agent.custom_tool_use":
            # Custom tool invocation - session is now idle
            print(f"\nCustom tool call: {event.name}")
            print(f"Input: {json.dumps(event.input)}")
            # Send result back (see below)
        elif event.type == "session.status_idle":
            print("\n--- Agent idle ---")
        elif event.type == "session.status_terminated":
            print("\n--- Session terminated ---")
            break

Provide Custom Tool Result / 提供自定义工具结果

client.beta.sessions.events.send(
    session_id=session.id,
    events=[
        {
            "type": "user.custom_tool_result",
            "custom_tool_use_id": "sevt_abc123",
            "content": [{"type": "text", "text": "All 42 tests passed."}],
        }
    ],
)

Poll Events / 轮询事件

events = client.beta.sessions.events.list(
    session_id=session.id,
)
for event in events.data:
    print(f"{event.type}: {event.id}")

Warning: Prefer the SDK over raw requests/httpx. If you hand-roll a poll loop, don't assume timeout=(5, 60) or httpx.Timeout(120) caps total call duration - both are per-chunk read timeouts (reset on every byte), so a trickling response can block forever. For a hard wall-clock deadline, track time.monotonic() at the loop level and bail explicitly, or wrap with asyncio.wait_for(). See Receiving Events.

警告：优先使用 SDK 而非原生 requests/httpx。 如果你自己手写轮询循环，不要以为 timeout=(5, 60) 或 httpx.Timeout(120) 限制了整个调用的总时长 —— 两者都是按数据块（per-chunk） 计的读取超时（每收到一个字节就重置），因此一个缓慢滴流的响应可能永远阻塞。若需要硬性的挂钟时间期限，请在循环层面跟踪 time.monotonic() 并显式退出，或用 asyncio.wait_for() 包裹。见 Receiving Events。

Full Streaming Loop with Custom Tools / 带自定义工具的完整流式循环

import json


def run_custom_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a custom tool and return the result."""
    if tool_name == "run_tests":
        # Your tool implementation here
        return "All tests passed."
    return f"Unknown tool: {tool_name}"


def run_session(client, session_id: str):
    """Stream events and handle custom tool calls."""
    while True:
        with client.beta.sessions.events.stream(
            session_id=session_id,
        ) as stream:
            tool_calls = []
            for event in stream:
                if event.type == "agent.message":
                    for block in event.content:
                        if block.type == "text":
                            print(block.text, end="", flush=True)
                elif event.type == "agent.custom_tool_use":
                    tool_calls.append(event)
                elif event.type == "session.status_idle":
                    break
                elif event.type == "session.status_terminated":
                    return

        if not tool_calls:
            break

        # Process custom tool calls
        results = []
        for call in tool_calls:
            result = run_custom_tool(call.name, call.input)
            results.append({
                "type": "user.custom_tool_result",
                "custom_tool_use_id": call.id,
                "content": [{"type": "text", "text": result}],
            })

        client.beta.sessions.events.send(
            session_id=session_id,
            events=results,
        )

Upload a File / 上传文件

with open("data.csv", "rb") as f:
    file = client.beta.files.upload(
        file=f,
    )

# Use in a session
session = client.beta.sessions.create(
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    environment_id=environment.id,
    resources=[{"type": "file", "file_id": file.id, "mount_path": "/workspace/data.csv"}],
)

List and Download Session Files / 列出并下载会话文件

List files the agent wrote to /mnt/session/outputs/ during a session, then download them.

列出智能体在会话期间写入 /mnt/session/outputs/ 的文件，然后下载它们。

# List files associated with a session
files = client.beta.files.list(
    scope_id=session.id,
    betas=["managed-agents-2026-04-01"],
)
for f in files.data:
    print(f.filename, f.size_bytes)
    # Download each file and save to disk
    file_content = client.beta.files.download(f.id)
    file_content.write_to_file(f.filename)

Tip: There's a brief indexing lag (~1-3s) between session.status_idle and output files appearing in files.list. Retry once or twice if the list is empty.

提示：从 session.status_idle 到输出文件出现在 files.list 之间有短暂的索引延迟（约 1-3 秒）。如果列表为空，可重试一到两次。

Session Management / 会话管理

# Get session details
session = client.beta.sessions.retrieve(session_id="sesn_011CZxAbc123Def456")
print(session.status, session.usage)

# List sessions
sessions = client.beta.sessions.list()

# Delete a session
client.beta.sessions.delete(session_id="sesn_011CZxAbc123Def456")

# Archive a session
client.beta.sessions.archive(session_id="sesn_011CZxAbc123Def456")

MCP Server Integration / MCP 服务器集成

# Agent declares MCP server (no auth here - auth goes in a vault)
agent = client.beta.agents.create(
    name="MCP Agent",
    model="claude-opus-5-5",
    mcp_servers=[
        {"type": "url", "name": "my-tools", "url": "https://my-mcp-server.example.com/sse"},
    ],
    tools=[
        {"type": "agent_toolset_20260401", "default_config": {"enabled": True}},
        {"type": "mcp_toolset", "mcp_server_name": "my-tools"},
    ],
)

# Session attaches vault(s) containing credentials for those MCP server URLs
session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    vault_ids=[vault.id],
)

See shared/managed-agents-tools.md §Vaults for creating vaults and adding credentials.

创建保管库和添加凭据见 shared/managed-agents-tools.md §Vaults。