订单客服 Agent

一个带两个 tool 的订单客服 Agent，为每个 Agent 框架各实现了一次。选择你的框架，下面每一节都会按这个选择显示。

SDK： Python · TypeScript

每个框架使用同一个订单客服 Agent。它有两个 tool，作用于一个内存中的 order desk：

Tool	Policy	Dex 做什么
`look_up_order`	读	无需审批即可运行，失败时重试。
`cancel_order`	非幂等写	在 `AwaitToolApproval` 中等待，然后最多运行一次。

cancel_order 只在订单仍处于处理中时才能取消它。

Python Agent 位于 examples/python/agent-harnesses。TypeScript Agent 位于 examples/typescript/src/products/agent-harnesses。

Agent 和它的 tool

Python 中的 Flow type 是 ClaudeAgentSDKOrderSupport，TypeScript 中是 ClaudeAgentSdkOrderSupportAgent。完整示例位于 examples/python/agent-harnesses 和 examples/typescript/src/products/agent-harnesses。

这些 tool 是 Claude Agent SDK tool。options 给出模型、system prompt 和 CLI 环境。tool policy 用每个 tool 自己的名字指定它。

claude_agent_sdk_order_support.pyPython
12345678910111213141516171819202122232425262728293031323334353637383940414243444546
def build_claude_agent_sdk_order_support_flow(
    order_desk: OrderDesk, model: str, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is a Claude model name, such as "claude-opus-5-5"."""

    # The tools are closures, so they call the OrderDesk passed in here.
    @tool(
        "look_up_order",
        "Look up one order by its order ID and return its items and status.",
        {"order_id": str},
    )
    async def look_up_order(arguments: dict[str, Any]) -> dict[str, Any]:
        order = order_desk.describe_order(arguments["order_id"])
        return {"content": [{"type": "text", "text": order}]}

    @tool(
        "cancel_order",
        "Cancel one order that has not shipped yet.",
        {"order_id": str, "reason": str},
    )
    async def cancel_order(arguments: dict[str, Any]) -> dict[str, Any]:
        result = order_desk.cancel_order(arguments["order_id"], arguments["reason"])
        return {"content": [{"type": "text", "text": result}]}

    # Every session gets the same options, so the factory does not read its context.
    def build_options(context: ClaudeAgentOptionsContext) -> ClaudeAgentOptions:
        return ClaudeAgentOptions(
            model=model,
            env=build_claude_cli_environment(os.environ),
            system_prompt=ORDER_SUPPORT_INSTRUCTIONS,
            # No Claude Code built-in tools: the agent uses only the order tools.
            tools=[],
            # Settings files could approve a tool before Dex asks for approval.
            setting_sources=[],
        )

    return claude_agent_session_flow(
        build_options,
        [look_up_order, cancel_order],
        flow_type=CLAUDE_AGENT_SDK_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/claude_agent_sdk_order_support.py

claude-agent-sdk-order-support-flow.tsTypeScript
12345678910111213141516171819202122232425262728293031323334353637383940
export function createClaudeAgentSdkOrderSupportFlow(
  orderDesk: OrderDesk,
  model: string,
  cliEnvironment: Readonly<Record<string, string>>,
  slack: Connection | undefined,
): AgentSessionFlow {
  const lookUpOrder = tool(
    "look_up_order",
    "Look up one order by its order ID and return its items and status.",
    { order_id: z.string() },
    async ({ order_id }) => ({ content: [{ type: "text", text: await orderDesk.describeOrder(order_id) }] }),
  );
  const cancelOrder = tool(
    "cancel_order",
    "Cancel one order that has not shipped yet.",
    { order_id: z.string(), reason: z.string() },
    async ({ order_id, reason }) => ({ content: [{ type: "text", text: await orderDesk.cancelOrder(order_id, reason) }] }),
  );
  return createClaudeAgentSessionFlow(
    {
      options: {
        model,
        systemPrompt: orderSupportInstructions,
        // No Claude Code built-in tools: the agent uses only the order tools.
        tools: [],
        // Settings files could approve a tool before Dex asks for approval.
        settingSources: [],
        // Only the variables in cliEnvironment reach the CLI, so no other Claude credential can take over.
        env: { ...cliEnvironment },
      },
      customTools: [lookUpOrder, cancelOrder],
      toolPolicies: {
        look_up_order: { effect: "read" },
        cancel_order: { effect: "non_idempotent_write" },
      },
      toolSources: createSlackConnectorToolSources(slack),
    },
    claudeAgentSdkOrderSupportFlowType,
  );
}

例子: examples/typescript/src/products/agent-harnesses/claude-agent-sdk/claude-agent-sdk-order-support-flow.ts

这个示例关闭了 Claude Code 的所有内置 tool，所以 Agent 只使用订单 tool。它也不加载任何 settings 文件，所以 settings 文件无法抢在 Dex 请求审批之前批准 tool。

Flow type 是 GoogleADKOrderSupport。完整示例位于 examples/python/agent-harnesses。

这些 tool 是 ADK LlmAgent 上的普通 Python 函数。tool policy 按名字指定每个 tool。

google_adk_order_support.pyPython
1234567891011121314151617181920212223242526272829
def build_google_adk_order_support_flow(
    order_desk: OrderDesk, model: str | BaseLlm, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is a Gemini model name, such as "gemini-2.5-flash"."""

    # The tools are closures, so they call the OrderDesk passed in here.
    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)

    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 = LlmAgent(
        name="order_support",
        model=model,
        instruction=ORDER_SUPPORT_INSTRUCTIONS,
        tools=[look_up_order, cancel_order],
    )
    return adk_agent_session_flow(
        agent,
        flow_type=GOOGLE_ADK_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/google_adk_order_support.py

Flow type 是 LangChainOrderSupport。完整示例位于 examples/python/agent-harnesses。

这些 tool 是 LangChain tool。一个工厂函数用 DurableToolCallMiddleware 构建 Agent。只有 Worker 会调用这个工厂函数，所以 Client 进程不需要 provider key。tool policy 按名字指定每个 tool。

langchain_order_support.pyPython
123456789101112131415161718192021222324252627282930313233343536
def build_langchain_order_support_flow(
    order_desk: OrderDesk,
    model: str | BaseChatModel,
    slack_connection: slack.Connection | None,
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is a LangChain model name, such as "openai:gpt-5.2"."""

    # The tools are closures, so they call the OrderDesk passed in here.
    @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)

    @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)

    # Only a Worker calls the factory, so a Client process needs no provider key.
    def build_order_support_agent() -> Pregel[Any, Any, Any, Any]:
        return create_agent(
            model,
            tools=[look_up_order, cancel_order],
            system_prompt=ORDER_SUPPORT_INSTRUCTIONS,
            middleware=[DurableToolCallMiddleware()],
        )

    return langchain_agent_session_flow(
        build_order_support_agent,
        flow_type=LANGCHAIN_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/langchain_order_support.py

Flow type 是 MastraOrderSupportAgent。完整示例位于 examples/typescript/src/products/agent-harnesses。

这些 tool 是 Mastra Agent 上的 Mastra tool。tool policy 按名字指定每个 tool。

mastra-order-support-flow.tsTypeScript
1234567891011121314151617181920212223242526272829303132
export function createMastraOrderSupportFlow(
  orderDesk: OrderDesk,
  model: MastraModelConfig,
  slack: Connection | undefined,
): AgentSessionFlow {
  const lookUpOrder = createTool({
    id: "look_up_order",
    description: "Look up one order by its order ID and return its items and status.",
    inputSchema: z.object({ order_id: z.string() }),
    execute: async ({ order_id }) => orderDesk.describeOrder(order_id),
  });
  const cancelOrder = createTool({
    id: "cancel_order",
    description: "Cancel one order that has not shipped yet.",
    inputSchema: z.object({ order_id: z.string(), reason: z.string() }),
    execute: async ({ order_id, reason }) => orderDesk.cancelOrder(order_id, reason),
  });
  const agent = new Agent({
    id: "order-support",
    name: "OrderSupport",
    instructions: orderSupportInstructions,
    model,
    tools: { look_up_order: lookUpOrder, cancel_order: cancelOrder },
  });
  return createMastraAgentSessionFlow(agent, mastraOrderSupportFlowType, {
    toolPolicies: {
      look_up_order: { effect: "read" },
      cancel_order: { effect: "non_idempotent_write" },
    },
    toolSources: createSlackConnectorToolSources(slack),
  });
}

例子: examples/typescript/src/products/agent-harnesses/mastra/mastra-order-support-flow.ts

Python 中的 Flow type 是 OpenAIAgentsOrderSupport，TypeScript 中是 OpenAIAgentsOrderSupportAgent。完整示例位于 examples/python/agent-harnesses 和 examples/typescript/src/products/agent-harnesses。

这些 tool 是该 SDK 的 function tool。Agent 是一个普通的 SDK Agent。tool policy 按名字指定每个 tool。

openai_agents_order_support.pyPython
12345678910111213141516171819202122232425262728293031
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

openai-agents-order-support-flow.tsTypeScript
12345678910111213141516171819202122232425262728293031323334
export function createOpenAIAgentsOrderSupportFlow(
  orderDesk: OrderDesk,
  model: string | Model,
  slack: Connection | undefined,
): AgentSessionFlow {
  const lookUpOrder = tool({
    name: "look_up_order",
    description: "Look up one order by its order ID and return its items and status.",
    parameters: z.object({ order_id: z.string() }),
    // Without errorFunction: null, tool() turns every thrown error into model-visible text.
    errorFunction: null,
    execute: ({ order_id }) => orderDesk.describeOrder(order_id),
  });
  const cancelOrder = tool({
    name: "cancel_order",
    description: "Cancel one order that has not shipped yet.",
    parameters: z.object({ order_id: z.string(), reason: z.string() }),
    errorFunction: null,
    execute: ({ order_id, reason }) => orderDesk.cancelOrder(order_id, reason),
  });
  const agent = new Agent({
    name: "OrderSupport",
    instructions: orderSupportInstructions,
    model,
    tools: [lookUpOrder, cancelOrder],
  });
  return createOpenAIAgentsSessionFlow(agent, openAIAgentsOrderSupportFlowType, {
    toolPolicies: {
      look_up_order: { effect: "read" },
      cancel_order: { effect: "non_idempotent_write" },
    },
    toolSources: createSlackConnectorToolSources(slack),
  });
}

例子: examples/typescript/src/products/agent-harnesses/openai-agents/openai-agents-order-support-flow.ts

Flow type 是 PydanticAIOrderSupport。完整示例位于 examples/python/agent-harnesses。

这些 tool 是 Agent 上的普通 Pydantic AI tool。tool policy 按名字指定每个 tool。Agent 推迟了模型检查，所以 Client 进程不需要 provider key。

pydantic_ai_order_support.pyPython
123456789101112131415161718192021222324252627
def build_pydantic_ai_order_support_flow(
    order_desk: OrderDesk, model: str | Model, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is a Pydantic AI model name, such as "openai:gpt-5.2"."""
    # The model is resolved at the first run, so a Client process needs no provider key.
    agent = Agent(model, instructions=ORDER_SUPPORT_INSTRUCTIONS, defer_model_check=True)

    # The tools are closures, so they call the OrderDesk passed in here.
    @agent.tool_plain
    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)

    @agent.tool_plain
    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)

    return pydantic_ai_session_flow(
        agent,
        flow_type=PYDANTIC_AI_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/pydantic_ai_order_support.py

Flow type 是 VercelAiOrderSupportAgent。完整示例位于 examples/typescript/src/products/agent-harnesses。

定义里包含模型、instructions、AI SDK tool 和 tool policy。Dex 根据 policy 决定是否需要审批。

vercel-ai-order-support-flow.tsTypeScript
123456789101112131415161718192021222324252627282930
export function createVercelAiOrderSupportFlow(
  orderDesk: OrderDesk,
  model: LanguageModel,
  slack: Connection | undefined,
): AgentSessionFlow {
  return createVercelAiAgentSessionFlow(
    {
      model,
      instructions: orderSupportInstructions,
      tools: {
        look_up_order: tool({
          description: "Look up one order by its order ID and return its items and status.",
          inputSchema: z.object({ order_id: z.string() }),
          execute: ({ order_id }) => orderDesk.describeOrder(order_id),
        }),
        cancel_order: tool({
          description: "Cancel one order that has not shipped yet.",
          inputSchema: z.object({ order_id: z.string(), reason: z.string() }),
          execute: ({ order_id, reason }) => orderDesk.cancelOrder(order_id, reason),
        }),
      },
      toolPolicies: {
        look_up_order: { effect: "read" },
        cancel_order: { effect: "non_idempotent_write" },
      },
      toolSources: createSlackConnectorToolSources(slack),
    },
    vercelAiOrderSupportFlowType,
  );
}

例子: examples/typescript/src/products/agent-harnesses/vercel-ai/vercel-ai-order-support-flow.ts

Worker

一个 Worker 为示例中所有 harness 的 Session Flow 提供服务。在 Python 中，Agent Session 的 Step 是 async 的，所以 Registry 允许 async handler。每个启动 Session 或调用其 RPC 的进程，都要构建同一个 Session Flow。Registry、Worker 和 Client 的设置对每个框架都一样；每个标签页只展示该框架的 Session Flow 如何构建和注册。

app.pyPython
1234567891011121314151617
        # 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)
        )

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

app.pyPython
12345678
def _new_claude_agent_sdk_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    flow = build_claude_agent_sdk_order_support_flow(
        order_desk, config.claude_agent_sdk_model, slack_connection
    )
    return AgentHarnessExample("claude-agent-sdk", flow, order_desk)

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

registry.tsTypeScript
123456789
// One Slack connection for the agent examples; without it they offer no Slack tools.
const slack = await openSlackConnectionFromProject();
// ...
// One order desk for the four agent harness examples; each harness registers its own session Flow type.
const orderDesk = new OrderDesk();
// ...
const claudeAgentSdkOrderSupportFlow = hasAnthropicApiKey()
  ? createClaudeAgentSdkOrderSupportFlow(orderDesk, loadClaudeAgentSdkOrderSupportModel(), buildClaudeCliEnvironment(), slack)
  : undefined;

例子: examples/typescript/src/registry.ts

app.pyPython
123456789101112
def _new_google_adk_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    model = _google_adk_model(config.google_adk_model)
    flow = build_google_adk_order_support_flow(order_desk, model, slack_connection)
    return AgentHarnessExample("google-adk", flow, order_desk)

# ...

def _google_adk_model(model: str) -> str | BaseLlm:
    return GoogleADKDemoModel() if is_demo_model(model) else model

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

app.pyPython
123456789101112
def _new_langchain_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    model = _langchain_model(config.langchain_model)
    flow = build_langchain_order_support_flow(order_desk, model, slack_connection)
    return AgentHarnessExample("langchain", flow, order_desk)

# ...

def _langchain_model(model: str) -> str | BaseChatModel:
    return LangChainDemoChatModel() if is_demo_model(model) else model

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

registry.tsTypeScript
1234567
// One Slack connection for the agent examples; without it they offer no Slack tools.
const slack = await openSlackConnectionFromProject();
// ...
// One order desk for the four agent harness examples; each harness registers its own session Flow type.
const orderDesk = new OrderDesk();
// ...
const mastraOrderSupportFlow = createMastraOrderSupportFlow(orderDesk, loadMastraOrderSupportModel(), slack);

例子: examples/typescript/src/registry.ts

app.pyPython
123456789101112
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

registry.tsTypeScript
1234567
// One Slack connection for the agent examples; without it they offer no Slack tools.
const slack = await openSlackConnectionFromProject();
// ...
// One order desk for the four agent harness examples; each harness registers its own session Flow type.
const orderDesk = new OrderDesk();
// ...
const openAIAgentsOrderSupportFlow = createOpenAIAgentsOrderSupportFlow(orderDesk, loadOpenAIAgentsOrderSupportModel(), slack);

例子: examples/typescript/src/registry.ts

app.pyPython
123456789101112
def _new_pydantic_ai_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    model = _pydantic_ai_model(config.pydantic_ai_model)
    flow = build_pydantic_ai_order_support_flow(order_desk, model, slack_connection)
    return AgentHarnessExample("pydantic-ai", flow, order_desk)

# ...

def _pydantic_ai_model(model: str) -> str | PydanticAIModel:
    return build_pydantic_ai_demo_model() if is_demo_model(model) else model

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

registry.tsTypeScript
1234567
// One Slack connection for the agent examples; without it they offer no Slack tools.
const slack = await openSlackConnectionFromProject();
// ...
// One order desk for the four agent harness examples; each harness registers its own session Flow type.
const orderDesk = new OrderDesk();
// ...
const vercelAiOrderSupportFlow = createVercelAiOrderSupportFlow(orderDesk, loadVercelAiOrderSupportModel(), slack);

例子: examples/typescript/src/registry.ts

把 Session 接入你的应用列出了你自己的应用需要的各个部分，每个部分一行说明。这段代码中的 blob cache 是存放大型 Flow 值的本地缓存；Registry 和 blob cache介绍了它。

Session 路由

每个 harness 提供相同的 HTTP 路由，你可以把它们放在一起对比试用。每个路由对应 Session Flow 上的一次调用：启动 Session、发送消息，或处理一次审批。

路由内部的调用是 Dex Agents API，在你的应用中也是同样的调用：

路由做什么	Python 中的 Dex Agents 调用	TypeScript 中的 Dex Agents 调用
启动 Session	`AgentSessionClient.start_session`	`startAgentSession`
发送用户消息	`AgentSessionClient.send_user_message`	`sendUserMessage` RPC，或 `sendAgentUserMessage`
读取 Session	`read_agent_event_stream_snapshot` 或 `AgentSessionClient.get_agent_snapshot`	`getAgentSnapshot` RPC，或 `readAgentSnapshot`
批准或拒绝 Tool Call	`AgentSessionClient.resolve_interrupt`	`resolveInterrupt` RPC

其余部分都属于示例本身：URL 路径、JSON 结构、Session token 和错误码。Python 和 TypeScript 示例在这些地方的设计略有不同。请围绕上面的调用设计你自己的路由。

启动 Session 会返回一个 Session token：用 DEX_AGENT_SESSION_KEY 对 Session ID 计算的 HMAC。其他每个路由都会检查 X-Agent-Session-Token header 中的 token，缺少 token 时返回 403。DEX_AGENT_SESSION_KEY 未设置时，示例会随机选一个 key，所以重启后所有 token 都会失效。真实的应用应改用自己的登录机制。