跳到主要内容
Super Durable
简体中文
English简体中文
GitHub登录注册 Cloud
菜单

OpenAI Agents SDK

查看 Markdown

OpenAI Agents SDK 用 instructions、function tool 和 handoff 定义 Agent。Dex adapter 把 SDK 的 runner 放进 durable session 中运行。它支持 Python 和 TypeScript。

何时使用​

Agent 已经用 OpenAI Agents SDK 编写时，无论是否使用 handoff，都可以使用它。保留 SDK 的 Agent 定义、tool 和 guardrail，由 Dex 管理 session。

Session 如何运行​

Adapter 会把 Agent graph 复制一次，并拦截每个 function tool。因此 SDK 在每次模型调用之后立即停下，返回模型请求的 tool call。每个 RunModel Step 调用一次模型。

每个 function tool call 在自己的 Step 中并行运行：read 或 idempotent write 用 CallTool，其他写操作用 CallSingleAttemptTool。需要审批的调用先在 AwaitToolApproval 中等待。下一个 RunModel 恢复 SDK 的 run state，并把记录下来的结果交给它。Tool 不会再次运行。

Handoff 不是 function tool。它在 RunModel 内运行，因此发生 handoff 的 Step 也会调用下一个 Agent 的模型。

查询订单的 turn 依次运行 RunModel、CallTool、ResumeTurn 和 RunModel。

安装​

把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。

pip install dex-agents-openai-agents

例子: agents/python/openai-agents/README.md

Example​

这个 example 就是每个 framework 页面共用的订单客服 Agent。

Flow type 是 OpenAIAgentsOrderSupport（Python），OpenAIAgentsOrderSupportAgent（TypeScript）。完整 example 位于 examples/python/agent-harnesses 和 examples/typescript/src/products/agent-harnesses。

Agent 与 tool​

Tool 是 SDK 的 function tool。Agent 是普通的 SDK Agent。Tool policy 指定每个 tool。

openai_agents_order_support.pyPython
def build_openai_agents_order_support_flow(
    order_desk: OrderDesk, model: str | Model, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is an OpenAI model name, such as "gpt-5.2"."""

    # The tools are closures, so they call the OrderDesk passed in here.
    @function_tool
    def look_up_order(order_id: str) -> str:
        """Look up one order by its order ID and return its items and status."""
        return order_desk.describe_order(order_id)

    @function_tool
    def cancel_order(order_id: str, reason: str) -> str:
        """Cancel one order that has not shipped yet."""
        return order_desk.cancel_order(order_id, reason)

    agent = Agent(
        name="OrderSupport",
        instructions=ORDER_SUPPORT_INSTRUCTIONS,
        model=model,
        tools=[look_up_order, cancel_order],
    )
    return openai_agents_session_flow(
        agent,
        flow_type=OPENAI_AGENTS_ORDER_SUPPORT_FLOW_TYPE,
        tool_sources=build_slack_tool_sources(slack_connection),
        tool_policies={
            "look_up_order": ToolPolicy(ToolEffect.READ),
            "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE),
        },
    )

例子: examples/python/agent-harnesses/agent_harness_examples/openai_agents_order_support.py

Worker​

一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。

app.pyPython
        # Agent session Steps are async, so the Registry allows async handlers.
        self.registry = Registry(
            tuple(example.session_flow for example in self.examples), allow_async_handlers=True
        )
        config.blob_cache_dir.mkdir(parents=True, exist_ok=True)
        self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30))
        self.worker = AsyncWorker(
            self.registry,
            self.blob_cache,
            WorkerOptions(
                server_address=config.server_address,
                pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS),
            ),
        )
        self.client = AsyncClient(
            self.registry, self.blob_cache, ClientOptions(server_address=config.server_address)
        )

# ...

def _new_openai_agents_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    model = _openai_agents_model(config.openai_agents_model)
    flow = build_openai_agents_order_support_flow(order_desk, model, slack_connection)
    return AgentHarnessExample("openai-agents", flow, order_desk)

# ...

def _openai_agents_model(model: str) -> str | Model:
    return OpenAIAgentsDemoModel() if is_demo_model(model) else model

例子: examples/python/agent-harnesses/agent_harness_examples/app.py

Session route​

每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 Session route。

http_routes.pyPython
    @blueprint.post("/sessions")
    async def start_session() -> Response:
        # The server picks the session ID, so a caller cannot claim an existing session.
        session_id = f"{harness_name}-order-support-{uuid.uuid4()}"
        await sessions.start_session(session_id)
        session_token = sign_session_id(app_state.config.session_key, session_id)
        return jsonify(session_id=session_id, session_token=session_token)

    @blueprint.post("/messages")
    async def send_user_message() -> Response:
        body, session_id = await session_request()
        accepted = await sessions.send_user_message(session_id, _required_string(body, "text"))
        return jsonify(accepted=accepted)

    @blueprint.post("/interrupts/resolve")
    async def resolve_interrupt() -> Response:
        body, session_id = await session_request()
        interrupt_id = _required_string(body, "interrupt_id")
        if not isinstance(approved := body.get("approved"), bool):
            raise BadRequest("approved must be a boolean")
        rejection_message = body.get("rejection_message")
        if rejection_message is not None and not isinstance(rejection_message, str):
            raise BadRequest("rejection_message must be a string")
        accepted = await sessions.resolve_interrupt(
            session_id, interrupt_id, approved=approved, rejection_message=rejection_message
        )
        return jsonify(accepted=accepted)

    @blueprint.get("/snapshot")
    async def get_agent_snapshot() -> Response:
        session_id = request.args.get("session_id", "")
        if not session_id:
            raise BadRequest("session_id is required")
        authorize_session(session_id)
        try:
            snapshot = await read_agent_event_stream_snapshot(
                app_state.client, example.session_flow, session_id
            )
        except FlowNotActiveOrNotFoundError as error:
            raise NotFound("the agent session is missing or no longer retained") from error
        return jsonify(snapshot.to_json())

例子: examples/python/agent-harnesses/agent_harness_examples/http_routes.py

在本地运行​

启动 Dex 与 example​

在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。

cd examples/python/agent-harnesses
uv sync --locked
mkdir -p definitions
uv run --frozen dex-agents-session-graph --flow-type OpenAIAgentsOrderSupport \
  --permission order-support.operate --out definitions/openai-agents-sdk.json
dexcli dev --flow-rendering-dir ./definitions

例子: examples/python/agent-harnesses/README.md

dexcli dev 会一直在前台运行。在第二个终端中启动 example：

cd examples/python/agent-harnesses
uv run --frozen python -m agent_harness_examples

例子: examples/python/agent-harnesses/README.md

Python 应用监听 127.0.0.1:8081。TypeScript 应用监听 127.0.0.1:8080。Dex Web 位于 127.0.0.1:8802。

没有设置 DEX_OPENAI_AGENTS_MODEL 时，Python Agent 使用 mock/order-support，这是一个不需要 key 的本地 demo model。收到 "look up order 1001" 时它调用 look_up_order，收到 "cancel order 1002" 时它调用 cancel_order。TypeScript example 没有 demo model，因此需要 OPENAI_API_KEY。

变量默认值含义
DEX_OPENAI_AGENTS_MODELPython：mock/order-support。TypeScript：gpt-5-mini。OpenAI 模型名称，例如 gpt-5.2。
OPENAI_API_KEY无OpenAI API key，使用真实模型时需要。
DEX_AGENT_SESSION_KEY随机 key签发 session token 的 key。

调用 session​

启动一个 session，让 Agent 取消订单 1002，然后批准这次调用：

base=http://127.0.0.1:8081/products/agent-harnesses/openai-agents
session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}')
id=$(echo "$session" | jq -r .session_id)
token=$(echo "$session" | jq -r .session_token)
curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \
  -H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}"
until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \
  | jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done
curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \
  -H "X-Agent-Session-Token: $token" \
  -d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}"

例子: examples/python/agent-harnesses/README.md

循环会反复读取 snapshot，直到 Agent 请求运行 cancel_order。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。

在 Dex Web 中审批​

也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见 在 Dex Web 中审批。

连接 Connector​

Agent 可以把 Connector 的 operation 当作 tool 调用，见 把 Connector 用作 tool。这个 example 加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

提供 tool​

在 Python 中，安装 dex-agents 的 connectors extra，以及 Connector 包，例如 dex-connector-slack。在 TypeScript 中，安装 @superdurable/connectors-sdk 以及 Connector 包，例如 @superdurable/connector-slack。安装命令见 Connector tools。

Worker 从项目配置中打开 Slack connection support-slack。Python 的 connector_tool 或 TypeScript 的 connectorTool 把 operation 变成 tool，由 ConnectorToolSource 持有。没有这个 connection 时，agent 不会得到 Slack tool。

slack_connector_tools.pyPython
def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]:
    """Returns the Slack tool source, or no source when Slack is not configured."""
    if connection is None:
        return []
    return [
        ConnectorToolSource([
            # The manifest does not ask for approval; this agent asks before every post.
            connector_tool(
                connection.post_thread_reply(), connection.reference, requires_approval=True
            ),
        ]),
    ]

# ...

async def open_slack_project_connection(
    environment: Mapping[str, str] = os.environ,
) -> SlackProjectConnection | None:
    # ...
    # An incomplete or invalid configuration raises here, so the Worker fails at startup.
    try:
        project = await load_from_environment(environment)
    except FileNotFoundError:
        _log_slack_not_set_up("the connector configuration file does not exist")
        return None
    # ...
    try:
        connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME)
    except BaseException:
        await project.close()
        raise
    return SlackProjectConnection(project, connection)

例子: examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py

Worker 在启动时打开一次 connection，每个 agent 都使用它：

__main__.pyPython
async def main() -> None:
    config = AgentHarnessExamplesConfig.from_env()
    slack_setup = await open_slack_project_connection()
    try:
        app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None)
        # ...
    finally:
        if slack_setup is not None:
            await slack_setup.aclose()

例子: examples/python/agent-harnesses/agent_harness_examples/__main__.py

Agent 把 tool source 和自己的 tool 一起传入：

openai_agents_order_support.pyPython
    return openai_agents_session_flow(
        agent,
        flow_type=OPENAI_AGENTS_ORDER_SUPPORT_FLOW_TYPE,
        tool_sources=build_slack_tool_sources(slack_connection),
        # ...
    )

例子: examples/python/agent-harnesses/agent_harness_examples/openai_agents_order_support.py

调用如何运行​

每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。原因见 Slack Connector tool。

Graph 中的每个 agent 都在自己的 tool 之后提供 Connector tool。在 Python 中，adapter 为 source 缓存的 listing（它的 cached_tools()）中的每个 tool 构建一个 function tool；ConnectorToolSource 会缓存它的整个 listing。在 TypeScript 中，它们是 non-strict function tool，tool 的 needsApproval 和 input guardrail 对它们不起作用。调用从不在 SDK 内运行；审批由 source 决定。

配置 connection​

Worker 从哪里找到 connection support-slack、Slack app 需要什么，见 Connection。

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.

这次发帖和 cancel_order 一样，会先等待审批。

限制​

  • 只有 function tool 拥有自己的 Step。Hosted tool，例如 web search，在模型调用期间由 provider 运行。RunModel 重跑时它们会重复执行。
  • SDK 自己运行的 tool，例如 MCP server tool 和本地 shell tool，无法等待审批。Adapter 默认拒绝它们。把写操作包装成 function tool，它就有了自己的 Step。
  • Function tool 的名称在整个 Agent graph 中必须唯一。Graph 在构建 Flow 时被复制，之后对它的修改不会影响 Flow。
  • Session 就是历史记录。不要使用 server 管理的 conversation state 或 SDK session。
  • 在 Python 中，作为 tool 使用的 Agent 会在一个 tool Step 中运行整个嵌套 run。TypeScript 会拒绝它；请改用 handoff。

Adapter README 列出了全部限制。Server 的限制参见 Limits for long sessions。