OpenAI Agents API

from openai import OpenAI

with OpenAI() as client:
    with client.beta.agents.sessions.create(
        agent={
            "model": "gpt-6-astra",
            "instructions": "写干净的代码，运行它，并报告真实的输出。",
        },
        environment={"type": "openai_hosted"},  # 由 OpenAI 开通并管理沙箱
        input="创建 tree.py，打印当前目录的文件树。运行它，并把输出给我看。",
        stream=True,
    ).with_result_collection() as stream:
        for event in stream:
            print(event.to_json(indent=None), flush=True)  # 实时事件
        result = stream.get_final_result()
    print(result.output_text)
    session_id = result.session_id  # 保存起来，后续轮次要用

from uuid import uuid4

from openai import OpenAI


def send_message(client: OpenAI, session_id: str, text: str, submission_key: str) -> None:
    client.beta.agents.sessions.events.create(
        session_id,
        idempotency_key=submission_key,  # 重试时复用同一个 key，避免重复提交
        events=[
            {
                "type": "agent.session.input.message",
                "input": [{"role": "user", "content": [{"type": "input_text", "text": text}]}],
            }
        ],
    )


submission_key = str(uuid4())  # 官方建议：先把 key 和消息一起保存，再提交

from openai import OpenAI

client = OpenAI()

# 1) 创建一次，得到 agent.id（可以把它当作“岗位模板”）
agent = client.beta.agents.create(
    model="gpt-6-astra",
    instructions="准确回答技术问题。",
    reasoning={"summary": "auto"},
)

# 2) 每次开新 session 时，通过 agent_id 引用它
session = client.beta.agents.sessions.create(
    agent_id=agent.id,
    environment={"type": "none"},
    input="解释一下 agent 是怎么连接 MCP 服务器的。",
)
print(session.id)

from openai import OpenAI
from openai.lib.beta.agents import function_tool

client = OpenAI()


@function_tool(name="get_weather", description="查询城市今天的天气")
def get_weather(city: str) -> str:
    return {"北京": "晴，25°C", "上海": "小雨，22°C"}.get(city, "暂无数据")


with client.beta.agents.sessions.create(
    agent={"model": "gpt-6-astra", "tools": [get_weather.definition]},  # 声明工具
    environment={"type": "none"},
    input="北京今天适合跑步吗？",
    stream=True,
    tool_handlers={get_weather.name: get_weather},  # 本地处理函数：迭代流时自动执行
) as stream:
    print(stream.get_final_result().output_text)

from openai import OpenAI

client = OpenAI()

with client.beta.agents.sessions.create(
    agent={
        "model": "gpt-6-astra",
        "instructions": "把每个版本说明分别委派给一个 subagent，"
        "提取对用户可见的变化和迁移步骤，等两边都完成后合并成一份摘要。",
        "multi_agent": {"enabled": True, "max_concurrent_subagents": 2},
    },
    environment={"type": "none"},
    input="版本 A：搜索支持按日期过滤。版本 B：导出接口改为返回下载 URL。",
    stream=True,
).with_result_collection() as stream:
    for _event in stream:
        pass
    result = stream.get_final_result()
print(result.output_text)