Claude Managed Agents

from anthropic import Anthropic

client = Anthropic()

# 1) Agent：只创建一次，保存 agent.id，之后每次开会话都引用它
agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-opus-5-5",
    system="你是一个编程助手，写干净、有文档的代码。",
    tools=[{"type": "agent_toolset_20260401"}],  # 内置工具集：bash、read、write、edit、glob、grep、web
)

# 2) Environment：云端容器，网络受限，但允许访问包管理器
environment = client.beta.environments.create(
    name="quickstart-env",
    config={"type": "cloud", "networking": {"type": "limited", "allow_package_managers": True}},
)

# 3) Session：引用 agent 和 environment
session = client.beta.sessions.create(
    agent=agent.id, environment_id=environment.id, title="Quickstart session"
)

# 4) 先打开事件流，再发送消息（stream-first，避免漏掉早期事件）
with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "写一个 Python 脚本，生成前 20 个斐波那契数并保存到 fibonacci.txt"}],
            }
        ],
    )
    for event in stream:
        if event.type == "agent.message":
            for block in event.content:
                if block.type == "text":
                    print(block.text, end="")
        elif event.type == "agent.tool_use":
            print(f"\n[使用工具：{event.name}]")
        elif event.type == "session.status_idle":
            print("\n\nAgent 完成。")
            break

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Code Reviewer",
    model="claude-opus-5-5",
    system="你是资深代码审查者。",
    tools=[
        {"type": "agent_toolset_20260401"},
        {
            "type": "custom",
            "name": "run_tests",
            "description": "运行测试套件",
            "input_schema": {
                "type": "object",
                "properties": {"test_path": {"type": "string", "description": "测试文件路径"}},
                "required": ["test_path"],
            },
        },
    ],
)


def run_custom_tool(name: str, tool_input: dict) -> str:
    return "All 42 tests passed." if name == "run_tests" else f"Unknown tool: {name}"


def run_session(session_id: str) -> None:
    while True:
        pending = []
        with client.beta.sessions.events.stream(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":
                    pending.append(event)  # 收集需要我们执行的调用
                elif event.type == "session.status_idle":
                    break
                elif event.type == "session.status_terminated":
                    return
        if not pending:
            return
        client.beta.sessions.events.send(
            session_id,
            events=[
                {
                    "type": "user.custom_tool_result",
                    "custom_tool_use_id": call.id,
                    "content": [{"type": "text", "text": run_custom_tool(call.name, call.input)}],
                }
                for call in pending
            ],
        )