Skip to main content
Super Durable
Menu

Add Dex to Your Agent

View as Markdown

Move the example agent of the Agent Quick Start into your own code: install your framework's adapter, copy the complete files, and serve the session from a web app.

What changes in your app

  • Your agent definition stays. You wrap it and give each tool a policy.
  • Your request handler changes shape. It sends the user's message to the session and returns. The reply arrives on the session's event stream, or in its snapshot.
  • You run two more things. A Worker process runs the session Steps, and Dex Server stores them. Dex Server runs in Dex Cloud or on your own infrastructure; Choose How to Run Dex compares the two. Dex is in Beta, Dex Server is at version 0.1, and one active node serves a cluster.
  • Each turn waits on Dex commits. A turn with three parallel tool calls adds about 0.4 seconds; see Latency and cost.

Run Agents in Production lists what to set up. Performance gives the numbers, and Limits gives the license.

What changes in your code

The example writes the agent as your framework alone would, then wraps it in a Dex agent session and gives each tool a policy. For these two function tools, that is the only change:

claude_agent_sdk_quick_start.pyPython
def build_plain_agent(order_desk: OrderDesk, model: str) -> ClaudeAgentOptions:
    """Returns the agent's options as the Claude Agent SDK alone writes them."""
    orders = create_sdk_mcp_server("orders", tools=build_order_tools(order_desk))
    return ClaudeAgentOptions(
        model=model,
        system_prompt=ORDER_SUPPORT_INSTRUCTIONS,
        mcp_servers={"orders": orders},
        allowed_tools=["mcp__orders__look_up_order", "mcp__orders__cancel_order"],
    )

def build_durable_agent(order_desk: OrderDesk, model: str) -> AgentSessionFlow:
    """Returns the same agent as a durable Dex agent session."""

    # Dex passes the tools to the CLI itself and decides when each one runs.
    def build_options(context: ClaudeAgentOptionsContext) -> ClaudeAgentOptions:
        return ClaudeAgentOptions(
            model=model,
            env=build_claude_cli_environment(os.environ),
            system_prompt=ORDER_SUPPORT_INSTRUCTIONS,
            tools=[],
            setting_sources=[],
        )

    return claude_agent_session_flow(
        build_options,
        build_order_tools(order_desk),
        flow_type="ClaudeAgentSDKQuickStart",
        tool_policies={
            "look_up_order": ToolPolicy(ToolEffect.READ),
            "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE),
        },
    )

Example: examples/python/agent-harnesses/agent_harness_examples/quick_start/claude_agent_sdk_quick_start.py

The Claude Agent SDK takes the tools in a list next to the options, so Dex runs each one itself.

Tools the framework runs itself, such as MCP servers and hosted tools, need the changes in your framework's Limits. The session runs in a Worker and is driven by a Client; Add Dex to Your Agent shows that host code.

Use it in your own agent

Install the adapter with its framework. Each adapter brings in the Dex Agents core. Then copy the complete files below.

pip install dex-agents-claude-agent-sdk

Example: agents/python/claude-agent-sdk

Authentication

The Worker authenticates the Claude Code CLI with ANTHROPIC_API_KEY. Set it in the Worker's environment. It never enters Flow state.

The example builds the CLI's environment so that ANTHROPIC_API_KEY is its only credential. Without the key, the example does not register the Claude agent. It serves the other harnesses.

The complete files

The quick start program is three files per language. They use only Dex, Dex Agents, and your framework, apart from the example's order desk with its tools and the settings the entry point reads. Read them in the repository:

FilePythonTypeScript
Your agent module, plain and durable, with every importclaude_agent_sdk · google_adk · langchain · openai_agents · pydantic_aiclaude-agent-sdk · mastra · openai-agents · vercel-ai
The runner: Registry, Worker, Client, and one turn with approvalssession.pyquick-start-session.ts
The entry point that asks for approval at the prompt__main__.pyquick-start-main.ts

Wire the session into your app

The change to your agent is small: the wrap in step 2 of the Agent Quick Start. The rest is hosting code that every Dex app has, a Worker that runs Steps and a Client that calls them. This function is the smallest host. It runs one Worker and one Client in your process, starts a session, sends one message, and returns the session after the turn:

embed.pyPython
"""The smallest host for a durable agent: one Worker, one Client, one turn."""

from __future__ import annotations

import asyncio

from dex import (
    AsyncClient,
    AsyncWorker,
    BlobCacheConfig,
    ClientOptions,
    Registry,
    WorkerOptions,
    open_blob_cache,
)
from dex_agents import AgentSessionClient, AgentSessionFlow, AgentSessionStatus, AgentSnapshot

async def ask_agent(
    session_flow: AgentSessionFlow,
    session_id: str,
    message: str,
    server_address: str = "127.0.0.1:8801",
    blob_cache_dir: str = ".dex-blob-cache",
) -> AgentSnapshot:
    """Starts a session, sends one message, and returns the session after the turn."""
    registry = Registry((session_flow,), allow_async_handlers=True)
    cache = open_blob_cache(BlobCacheConfig(blob_cache_dir, 1 << 30))
    worker = AsyncWorker(registry, cache, WorkerOptions(server_address=server_address))
    client = AsyncClient(registry, cache, ClientOptions(server_address=server_address))
    worker_task = asyncio.create_task(worker.start())
    try:
        await worker.wait_pull_registered(timeout=30)
        sessions = AgentSessionClient(client, session_flow)
        await sessions.start_session(session_id)
        await sessions.send_user_message(session_id, message)
        while True:
            snapshot = await sessions.get_agent_snapshot(session_id)
            state = snapshot.state
            if state and state.turn_count >= 1:
                if state.status == AgentSessionStatus.WAITING_FOR_USER_MESSAGE:
                    return snapshot
            if state and state.status == AgentSessionStatus.FAILED:
                raise RuntimeError(f"the session failed: {state.last_turn_error}")
            await asyncio.sleep(0.5)
    finally:
        await client.close()
        await worker.close()
        await worker_task
        cache.close()

Example: examples/python/agent-harnesses/agent_harness_examples/quick_start/embed.py

It takes the session Flow that your module's build_durable_agent or buildDurableAgent returns. A turn that needs an approval waits until your app resolves it; Approve from your application shows how.

A web app splits the same pieces across routes:

  1. A Registry with your session Flow, a Worker that runs its Steps, and a Client that calls it, as in the function above. The Worker connects out to Dex, so it needs no open port. Both open a blob cache, a local cache for large values; Registry and blob cache explains it.
  2. A route that starts a session. The session ID is the Flow ID.
  3. A route that sends a user message, and one that reads the session snapshot with the agent's reply. To stream the reply as it is written, read the AgentEvents Stream, as the next section shows.
  4. A route that approves or rejects a waiting tool call. Approve from your application covers it.

The Order Support Agent example has these routes over HTTP, for every framework.

What replaces the framework call in a web handler

The handler queues the message on the session and returns.

This replaces the query call that waited for the reply.

The reply arrives on the session's AgentEvents Stream, which your UI reads as server-sent events. The function above polls the session snapshot, which suits scripts and tests. A chat UI reads the stream instead.

This complete app hosts the Worker and the Client and serves every route a chat UI needs: start a session, send a message, read the session, stream its events, and answer an approval. The whole module, quick_start/web.py:

web.pyPython
"""A complete web app for a durable agent: start, message, read, stream, approve.

It hosts the Worker and the Client in the web process. Add your own
authentication before you expose these routes: anyone who knows a session ID
can read and change that session here.
"""

from __future__ import annotations

import uuid
from pathlib import Path
from typing import Any

from dex_agents import AgentSessionFlow
from dex_agents.ui_stream import (
    AGENT_EVENT_STREAM_HEADERS,
    agent_snapshot_to_json,
    stream_agent_event_frames,
)
from quart import Quart, Response, jsonify, make_response, request

from agent_harness_examples.quick_start.safe_send import send_user_message_once
from agent_harness_examples.quick_start.session import QuickStartApp

def create_web_app(
    session_flow: AgentSessionFlow, server_address: str, blob_cache_dir: Path
) -> Quart:
    """Returns a Quart app whose routes run one durable agent's sessions."""
    web = Quart(__name__)
    apps: list[QuickStartApp] = []

Example: examples/python/agent-harnesses/agent_harness_examples/quick_start/web.py

Next

Sessions