--- title: "Order Support Agent" description: "An order support agent with two tools, built once for each agent framework. Pick your framework, and every section below follows that choice." canonical_url: "https://docs.superdurable.io/use-cases/order-support-agent/" source: "docs/content/use-cases/order-support-agent.mdx" locale: "en" framework: "claude-agent-sdk" language: "python" --- > Index of every page for AI agents: https://docs.superdurable.io/llms.txt # Order Support Agent This file shows the Claude Agent SDK version. Other frameworks: [Google ADK](https://docs.superdurable.io/use-cases/order-support-agent.google-adk.md) · [LangChain and LangGraph](https://docs.superdurable.io/use-cases/order-support-agent.langchain-langgraph.md) · [Mastra](https://docs.superdurable.io/use-cases/order-support-agent.mastra.md) · [OpenAI Agents SDK](https://docs.superdurable.io/use-cases/order-support-agent.openai-agents-sdk.md) · [Pydantic AI](https://docs.superdurable.io/use-cases/order-support-agent.pydantic-ai.md) · [Vercel AI SDK](https://docs.superdurable.io/use-cases/order-support-agent.vercel-ai-sdk.md) Code on this page is in Python. Other languages: [TypeScript](https://docs.superdurable.io/use-cases/order-support-agent.typescript.md) An order support agent with two tools, built once for each agent framework. Pick your framework, and every section below follows that choice. **SDKs:** Python · TypeScript Every framework uses the same order support agent. It has two tools on an in-memory order desk: | Tool | Policy | What Dex does | | --------------- | -------------------- | ----------------------------------------------------- | | `look_up_order` | read | Runs without approval and retries on failure. | | `cancel_order` | non-idempotent write | Waits in `AwaitToolApproval`, then runs at most once. | `cancel_order` cancels an order only while it is still processing. The Python agents are under [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses). The TypeScript agents are under [examples/typescript/src/products/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/agent-harnesses). ## The agent and its tools The Flow type is `ClaudeAgentSDKOrderSupport` in Python and `ClaudeAgentSdkOrderSupportAgent` in TypeScript. The full example is under [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) and [examples/typescript/src/products/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/agent-harnesses). The tools are Claude Agent SDK tools. The options give the model, the system prompt, and the CLI environment. The tool policies name each tool by its own name. **Python** ```python 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), }, ) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py) The example turns off every built-in Claude Code tool, so the agent uses only the order tools. It also loads no settings files, so no settings file can approve a tool before Dex asks. ## Worker One Worker serves the session Flows of every harness in the example. In Python, agent session Steps are async, so the Registry allows async handlers. Build the same session Flow in every process that starts sessions or calls their RPCs. The Registry, Worker, and Client setup is the same for every framework; each tab shows only how that framework's session Flow is built and registered. **Python** ```python # 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) ) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/app.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) **Python** ```python 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) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/app.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) [Wire the session into your app](https://docs.superdurable.io/agent-frameworks/add-to-your-agent.md#wire-it-into-your-app) lists the pieces your own app needs, with one line on each. The blob cache in this code is a local cache for large Flow values; [Registry and blob cache](https://docs.superdurable.io/components/application.md#blob-cache) explains it. ## Session routes Every harness serves the same HTTP routes, so you can try them side by side. Each route is one call on the session Flow: start a session, send a message, or resolve an approval. The calls inside the routes are the Dex Agents API, and they are the same in your app: | What the route does | Dex Agents call in Python | Dex Agents call in TypeScript | | ----------------------------- | ----------------------------------------------------------------------------- | ---------------------------------------------------- | | Start a session | `AgentSessionClient.start_session` | `startAgentSession` | | Send a user message | `AgentSessionClient.send_user_message` | the `sendUserMessage` RPC, or `sendAgentUserMessage` | | Read the session | `read_agent_event_stream_snapshot` or `AgentSessionClient.get_agent_snapshot` | the `getAgentSnapshot` RPC, or `readAgentSnapshot` | | Approve or reject a tool call | `AgentSessionClient.resolve_interrupt` | the `resolveInterrupt` RPC | Everything else belongs to the example: the URL paths, the JSON shapes, the session token, and the error codes. The Python and TypeScript examples shape them a little differently. Design your own routes around the calls above. Starting a session returns a session token: an HMAC of the session ID under `DEX_AGENT_SESSION_KEY`. Every other route checks the token in the **X-Agent-Session-Token** header and answers 403 without it. Without `DEX_AGENT_SESSION_KEY`, the example picks a random key, so a restart ends every token. A real application uses its own sign-in instead. Every harness serves these same routes. [Session routes](https://docs.superdurable.io/use-cases/order-support-agent.md#session-routes) explains them and the session token. **Python** ```python @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()) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/http_routes.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/http_routes.py) ## Run it locally In a first terminal, render the session graph, then start Dex. Dex Web reads the graph when it starts, to show the session and its approval controls. **Python** ```bash cd examples/python/agent-harnesses uv sync --locked mkdir -p definitions uv run --frozen dex-agents-session-graph --flow-type ClaudeAgentSDKOrderSupport \ --permission order-support.operate --out definitions/claude-agent-sdk.json dexcli dev --flow-rendering-dir ./definitions ``` Example: [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) **dexcli dev** stays in the foreground. In a second terminal, start the example: **Python** ```bash cd examples/python/agent-harnesses ANTHROPIC_API_KEY=... uv run --frozen python -m agent_harness_examples ``` Example: [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) The Python app serves on **127.0.0.1:8081**. The TypeScript app serves on **127.0.0.1:8080**. Dex Web is at `http://127.0.0.1:8802`. | Variable | Default | Meaning | | ---------------------------- | ------------------- | ------------------------------------------------------------------------------------------- | | `ANTHROPIC_API_KEY` | none; required | The Anthropic API key of the Worker. | | `DEX_CLAUDE_AGENT_SDK_MODEL` | **claude-opus-5-5** | The Claude model name. | | `DEX_AGENT_SESSION_KEY` | a random key | The key that signs session tokens. A random key ends every token when the process restarts. | Start a session, ask the agent to cancel order 1002, and approve the call: **Python** ```bash base=http://127.0.0.1:8081/products/agent-harnesses/claude-agent-sdk 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}" ``` Example: [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) The loop reads the snapshot until the agent asks to run `cancel_order`. That call waits for approval, then runs once. Read the snapshot again to see the agent's answer. ## Next - [Agent Quick Start](https://docs.superdurable.io/agent-frameworks/quick-start.md) - [Approvals](https://docs.superdurable.io/agent-frameworks/approvals.md) - [Connectors as Tools](https://docs.superdurable.io/agent-frameworks/connectors-as-tools.md)