Dify Agent 官方示例实践：从 Pydantic AI 网关模型到 Run 事件轮询、同步客户端与 SSE 流式消费

2026-09-05 14:05:35作者：傅爽业Veleda

Build Agentic workflows, RAG pipelines, with rich AI model and tool support on one collaborative workspace. Deploy on cloud, VPC, or self-hosted, so teams move from prototype to production without rebuilding the stack.

项目地址：https://gitcode.com/GitHub_Trending/di/dify

本文基于 Dify Agent 示例文档，完整覆盖 dify-agent/examples/dify_agent/dify_agent_examples 目录下四个官方示例脚本的用途、前置依赖、环境变量与运行方式，并结合 Python 客户端源码与 LLM 适配层实现解释其底层调用链路。读完后你可以：在本机跑通“经 Dify API 网关调用大模型”的 Pydantic AI Agent、创建一次由 Dify 插件守护进程（plugin daemon）提供模型的 Run，并用游标轮询、同步等待、SSE 三种方式消费 Run 事件。

一、示例的定位与运行时依赖

官方文档开篇说明了这组示例的归属与隔离原因：这些示例位于 examples/dify_agent/dify_agent_examples，之所以与 Agenton 示例分开，是因为它们依赖 Dify Agent 运行时服务，包括 FastAPI 服务器、Redis 或插件守护进程（plugin daemon）。

四个示例脚本分别是：

示例	脚本	解决什么问题
Run a Dify plugin-daemon backed model	run_pydantic_ai_agent.py	不启动 Run 服务器，直接把 Dify API 的 LLM 网关当作 Pydantic AI 的 Provider 用
Poll run events	run_server_consumer.py	异步客户端创建 Run，并用游标（cursor）轮询事件
Use the synchronous client	run_server_sync_client.py	同步风格客户端，创建 Run 后阻塞等待终态
Stream run events with SSE	run_server_sse_consumer.py	用 SSE 流式接收已创建 Run 的事件

此外，示例包自带一个轻量 CLI（main.py）：直接执行 python -m dify_agent_examples 会列出全部可运行的模块（python -m dify_agent_examples.run_pydantic_ai_agent 等），使用 --copy-to DEST 可把示例文件复制到新目录作为起点。该 CLI 中 EXAMPLE_MODULES 元组（第 10-15 行）正是文档四个章节对应脚本的权威清单。

需要特别注意的是文档和脚本注释共同强调的一点：dify_agent_examples 包不是发布包的一部分（not part of the published package），这些示例设计为从源码检出（source checkout）运行，这也是单元测试只验证“四个模块能从仓库检出导入”的原因。

二、示例一：运行 Dify 插件守护进程提供模型的 Pydantic AI Agent

这是四个示例中唯一“绕过 Run 服务器”的路径：它通过 DifyApiLLMProvider 和 DifyLLMAdapterModel，把 Dify API 的 inner LLM 网关适配成 Pydantic AI 的模型接口，直接驱动一个 pydantic_ai.Agent 流式运行。

前置条件

脚本 docstring 列出的三个前置条件：

先同步服务器运行时依赖：uv sync --project dify-agent --extra server；
运行中的 Dify API 必须启用 inner Agent LLM 端点；
在 dify-agent/.env 中填写真实的租户、用户、应用、插件、提供方与模型。

完整代码

run_pydantic_ai_agent.py 的核心逻辑：

"""Run a Pydantic AI agent through the Dify API LLM gateway.

Prerequisites:
- Sync the server runtime dependencies first: `uv sync --project dify-agent --extra server`.
- Run the Dify API with its inner Agent LLM endpoint enabled.
- Fill `dify-agent/.env` with a real tenant, user, app, plugin, provider, and model.

This example is meant to be run from a source checkout because
`dify_agent_examples` is not part of the published package.

Example from the repository root:
    PYTHONPATH=dify-agent/src:dify-agent/examples/dify_agent \
    uv run --project dify-agent python -m dify_agent_examples.run_pydantic_ai_agent
"""

from __future__ import annotations

import asyncio
import os
from pathlib import Path
from uuid import uuid4

import httpx
from pydantic_ai import Agent

from dify_agent.adapters.llm import DifyApiLLMProvider, DifyLLMAdapterModel
from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig


PROJECT_ROOT = Path(__file__).resolve().parents[3]


def load_env_file(path: Path) -> None:
    """Load simple KEY=VALUE lines without adding a dotenv dependency."""
    if not path.exists():
        return

    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if value:
        return value
    raise RuntimeError(f"Missing required environment variable: {name}")


async def main() -> None:
    load_env_file(PROJECT_ROOT / ".env")

    async with httpx.AsyncClient(timeout=600, trust_env=False) as http_client:
        provider = DifyApiLLMProvider(
            plugin_id=required_env("DIFY_AGENT_PLUGIN_ID"),
            inner_api_url=required_env("DIFY_INNER_API_URL"),
            inner_api_key=required_env("DIFY_INNER_API_KEY"),
            execution_context=DifyExecutionContextLayerConfig(
                tenant_id=required_env("DIFY_AGENT_TENANT_ID"),
                user_id=required_env("DIFY_AGENT_USER_ID"),
                user_from="account",
                app_id=required_env("DIFY_AGENT_APP_ID"),
                agent_mode="single_step",
                invoke_from="debugger",
            ),
            agent_run_id=str(uuid4()),
            http_client=http_client,
        )
        model = DifyLLMAdapterModel(
            required_env("DIFY_AGENT_MODEL_NAME"),
            provider,
            model_provider=required_env("DIFY_AGENT_PROVIDER"),
        )
        agent = Agent(model=model)
        async with agent.run_stream("Explain the theory of relativity") as run:
            async for piece in run.stream_output():
                print(piece, end="", flush=True)
            print(run.usage)


if __name__ == "__main__":
    asyncio.run(main())

从仓库根目录运行的方式（脚本内 docstring 给出，PYTHONPATH 同时指向包源码目录与示例目录）：

PYTHONPATH=dify-agent/src:dify-agent/examples/dify_agent \
    uv run --project dify-agent python -m dify_agent_examples.run_pydantic_ai_agent

环境变量逐项说明

脚本自带极简 .env 解析器（load_env_file，不引入 dotenv 依赖），required_env 在缺失时直接抛 RuntimeError，因此以下 8 个变量缺一不可：

变量	用途
`DIFY_AGENT_PLUGIN_ID`	Dify 插件守护进程中的模型插件标识，如 `langgenius/openai`
`DIFY_INNER_API_URL`	Dify API 的 `/inner/api/...` 根地址
`DIFY_INNER_API_KEY`	发送给 Dify API inner 插件端点的密钥，需与 Dify API 的 `INNER_API_KEY_FOR_PLUGIN` 一致
`DIFY_AGENT_TENANT_ID`	执行上下文中的租户 ID
`DIFY_AGENT_USER_ID`	执行上下文中的用户 ID（`user_from="account"` 表示账号体系用户）
`DIFY_AGENT_APP_ID`	执行上下文中的应用 ID
`DIFY_AGENT_MODEL_NAME`	具体模型名，如 `gpt-4o-mini`
`DIFY_AGENT_PROVIDER`	模型提供方，如 `openai`

源码层面的实现要点

从 adapters/llm/model.py 的模块 docstring 可以看到这条链路的三个关键设计：

Agent 调用的是 Dify API 的可信 LLM 网关（trusted LLM gateway），而不是在本地托管各模型厂商 SDK——模型凭证由 Dify API 在调用时解析，这正是示例只需配置插件 ID、无需填模型 API key 的原因；
适配器把 Pydantic AI 的消息映射为网关兼容的 Graphon 请求与流式响应模式；
Pydantic AI 只保留 token 计数，适配器会针对单个模型实例的生命周期单独累加 Dify 完整的 Graphon usage，Runner 在每轮模型/工具交互结束后读取累计用量——这解释了示例末尾 print(run.usage) 能输出完整用量明细。

三、服务端公共前置：启动 Dify Agent Run 服务器

其余三个示例（轮询、同步、SSE）共享同一套运行时：一个 FastAPI 服务器 + Redis。Get started 文档给出的最小部署路径是：

安装依赖（运行 API 服务器只需要 server extra）：

cd dify-agent
uv sync --all-extras --all-groups   # 或最小化：uv sync --project dify-agent --extra server

准备 Redis（已有可跳过）：

docker run -d \
  --name dify-agent-redis \
  -p 6379:6379 \
  redis:7-alpine

在 dify-agent/.env 写入最小配置：

cat > .env <<'EOF'
DIFY_AGENT_REDIS_URL=redis://localhost:6379/0
DIFY_AGENT_REDIS_PREFIX=dify-agent

DIFY_AGENT_PLUGIN_DAEMON_URL=http://localhost:5002
DIFY_AGENT_PLUGIN_DAEMON_API_KEY=replace-with-plugin-daemon-server-key

DIFY_AGENT_INNER_API_URL=http://localhost:5001
DIFY_AGENT_INNER_API_KEY=replace-with-dify-inner-api-key-for-plugin
EOF

其中 DIFY_AGENT_PLUGIN_DAEMON_URL / DIFY_AGENT_PLUGIN_DAEMON_API_KEY 指向 Dify 插件守护进程（Docker 部署中通常复用 PLUGIN_DAEMON_KEY 的值）；DIFY_AGENT_INNER_API_URL / DIFY_AGENT_INNER_API_KEY 指向 Dify API 的 inner 端点（对应 Dify API 的 INNER_API_KEY_FOR_PLUGIN）。

启动服务器（对应 server/app.py 的 FastAPI 应用）：
```
# 开发模式（uvicorn reload）
uv run --project dify-agent uvicorn dify_agent.server.app:app --reload
# 或使用 Makefile
make dev
```
服务器默认监听 http://127.0.0.1:8000；ServerSettings 会从当前 dify-agent 目录（或从仓库根执行时的 dify-agent/.env）读取配置。

三个服务器示例共享同一个“运行组合”（Run Composition）结构，这是理解它们的前提。组合由若干 RunLayerSpec 层按声明顺序与 deps 依赖构成：

prompt 层：PLAIN_PROMPT_LAYER_TYPE_ID + PromptLayerConfig(prefix=..., user=...)，即系统前缀与用户提示词；
execution_context 层：DifyExecutionContextLayerConfig，携带 tenant_id / user_id / user_from / app_id / agent_mode / invoke_from，让 Dify API 按正确的产品上下文解析模型凭证；
模型层：名称固定为 DIFY_AGENT_MODEL_LAYER_ID，类型为 DIFY_PLUGIN_LLM_LAYER_TYPE_ID，通过 deps={"execution_context": "execution_context"} 声明对上层的依赖，配置为 DifyPluginLLMLayerConfig(plugin_id=..., model_provider=..., model=...)。

run_server_consumer.py 中还提供了一段注释掉的 plugin tools 层 示例，说明 API 调用方应传入准备好的参数与 JSON schema，而不是依赖 dify-agent 去拉取合并 daemon 声明：

# Minimal plugin-tools example. API callers should pass
# prepared parameters + JSON schema instead of relying on
# dify-agent to fetch and merge daemon declarations.
# from dify_agent.layers.dify_plugin import (
#     DifyPluginToolConfig,
#     DifyPluginToolParameter,
#     DifyPluginToolParameterForm,
#     DifyPluginToolParameterType,
#     DifyPluginToolsLayerConfig,
# )
# RunLayerSpec(
#     name="tools",
#     type="dify.plugin.tools",
#     deps={"execution_context": "execution_context"},
#     config=DifyPluginToolsLayerConfig(
#         tools=[
#             DifyPluginToolConfig(
#                 plugin_id="langgenius/search",
#                 provider="search",
#                 tool_name="web_search",
#                 credential_type="api-key",
#                 credentials={"api_key": "replace-with-tool-key"},
#                 runtime_parameters={"site": "docs.dify.ai"},
#                 parameters=[
#                     DifyPluginToolParameter(
#                         name="query",
#                         type=DifyPluginToolParameterType.STRING,
#                         form=DifyPluginToolParameterForm.LLM,
#                         required=True,
#                         llm_description="Search query",
#                     ),
#                 ],
#                 parameters_json_schema={
#                     "type": "object",
#                     "properties": {
#                         "query": {"type": "string", "description": "Search query"}
#                     },
#                     "required": ["query"],
#                 },
#             )
#         ]
#     ),
# )

run_server_sync_client.py 中包含同一段注释（第 66-108 行），此处不再重复。

四、示例二：创建 Run 并用游标轮询事件（异步客户端）

run_server_consumer.py 展示异步 Client 的完整用法：create_run 创建运行，然后以 after=cursor 方式翻页拉取事件，直到出现 run_succeeded 或 run_failed。

API_BASE_URL = "http://localhost:8000"
TENANT_ID = "replace-with-tenant-id"
USER_ID = "replace-with-user-id"
APP_ID = "replace-with-app-id"
PLUGIN_ID = "langgenius/openai"
PLUGIN_PROVIDER = "openai"
MODEL_NAME = "gpt-4o-mini"


async def main() -> None:
    async with Client(base_url=API_BASE_URL) as client:
        run = await client.create_run(
            CreateRunRequest(
                composition=RunComposition(
                    layers=[
                        RunLayerSpec(
                            name="prompt",
                            type=PLAIN_PROMPT_LAYER_TYPE_ID,
                            config=PromptLayerConfig(
                                prefix="You are a concise assistant.",
                                user="Say hello from the Dify Agent API server example.",
                            ),
                        ),
                        RunLayerSpec(
                            name="execution_context",
                            type=DIFY_EXECUTION_CONTEXT_LAYER_TYPE_ID,
                            config=DifyExecutionContextLayerConfig(
                                tenant_id=TENANT_ID,
                                user_id=USER_ID,
                                user_from="account",
                                app_id=APP_ID,
                                agent_mode="workflow_run",
                                invoke_from="service-api",
                            ),
                        ),
                        RunLayerSpec(
                            name=DIFY_AGENT_MODEL_LAYER_ID,
                            type=DIFY_PLUGIN_LLM_LAYER_TYPE_ID,
                            deps={"execution_context": "execution_context"},
                            config=DifyPluginLLMLayerConfig(
                                plugin_id=PLUGIN_ID,
                                model_provider=PLUGIN_PROVIDER,
                                model=MODEL_NAME,
                            ),
                        ),
                        # 上文“plugin tools 层”注释示例，省略
                    ],
                ),
            )
        )
        print("created run", run)

        cursor = "0-0"
        while True:
            page = await client.get_events(run.run_id, after=cursor)
            cursor = page.next_cursor or cursor
            for event in page.events:
                print("event", event)
                if event.type in {"run_succeeded", "run_failed"}:
                    return
            await asyncio.sleep(0.5)

两个值得注意的客户端语义（docstring 与源码共同确认）：

Client.create_run 只做一次 POST 尝试，不重试。客户端在超时等情况下不确定请求是否已被接受，官方建议通过轮询或 SSE 重放（replay）来恢复，而不是盲目重发 POST /runs；
游标轮询从 "0-0" 开始，page.next_cursor 为空时保持原游标，配合 0.5 秒间隔实现轻量轮询，直到终态事件出现。

对照 client/_client.py 的实现：create_run（L350）、get_events(run_id, *, after="0-0", limit=100)（L489）与 stream_events（L637）都是默认参数与示例用法一致的，after 默认值 "0-0" 正是轮询的起始游标。

五、示例三：使用同步客户端并阻塞等待终态

run_server_sync_client.py 与异步版的差异集中在两处：

用同步 with Client(...) 上下文替代 async with；
不调用 get_events 手动轮询，而是调用 wait_run_sync 阻塞直到终态。

def main() -> None:
    with Client(base_url=API_BASE_URL) as client:
        run = client.create_run_sync(
            CreateRunRequest(
                composition=RunComposition(
                    layers=[
                        # 与异步示例完全相同的 prompt / execution_context / 模型层
                        # 三个 RunLayerSpec，此处省略重复结构，见上文第三节
                    ],
                ),
            )
        )
        print("created run", run)
        terminal = client.wait_run_sync(run.run_id, poll_interval_seconds=0.5)
        print("terminal status", terminal)

要点：

create_run_sync 对应 client/_client.py 中的同步封装，其 docstring 特别提醒：它不会重试 POST /runs——如果发生超时，应该检查服务器状态或显式新建 Run，而不是假设原请求未被接受；
wait_run_sync（L785）内部以 poll_interval_seconds=0.5 周期性查询，返回终态对象，适合脚本化、批处理等不需要细粒度事件流的场景；
该示例中执行上下文使用 agent_mode="workflow_run" 与 invoke_from="service-api"，与示例二一致，表示模拟工作流运行场景下的服务 API 调用来源。

六、示例四：用 SSE 流式消费 Run 事件

run_server_sse_consumer.py 是最短的示例，它不负责创建 Run，而是消费由 run_server_consumer.py（或任意 HTTP 客户端）创建、且服务器仍可用时的某个 Run：

API_BASE_URL = "http://localhost:8000"
RUN_ID = "replace-with-run-id"


async def main() -> None:
    async with Client(base_url=API_BASE_URL, stream_timeout=None) as client:
        async for event in client.stream_events(RUN_ID):
            print(event)

docstring 给出的三条语义，均值得在实际接入时留意：

Python 客户端会把 SSE 帧解析为带类型的协议事件（typed protocol events），async for 迭代得到的是结构化事件对象而非原始文本；
默认使用最新 event id 自动重连，因此 stream_timeout=None 配合重连可以维持长期流式消费；
畸形帧与 HTTP 4xx 响应会直接失败而不重连——4xx 属于客户端错误（如 Run 不存在），重连没有意义。

stream_events 的客户端实现在 client/_client.py，配套的异常类型（DifyAgentHTTPError、DifyAgentStreamError、DifyAgentTimeoutError、DifyAgentValidationError、DifyAgentNotFoundError 等）统一从 client/init.py 导出，便于调用方按错误类别做处理。

七、示例的可验证性与进一步阅读

四个示例模块的可导入性由 tests/local/examples/test_dify_agent_examples.py 守护：测试把 examples/dify_agent 前置到 sys.path，逐一 importlib.import_module 全部四个脚本，保证示例代码与包 API 不漂移；
服务器端行为（路由、鉴权、SSE、Redis 事件存储等）有更完整的本地测试覆盖，如 tests/local/dify_agent/server/ 下的 test_runs_routes.py、test_sse.py 等；
若要理解执行上下文的各层（dify.runtime、dify.shell、knowledge、ask-human 等）与运行时资源模型，可继续阅读 Get started、concepts 与 guide 文档，以及 dify-agent 包 README。

关键文件索引

文件	作用
docs/dify-agent/examples/index.md	本文依据的示例文档，四个示例的官方入口
run_pydantic_ai_agent.py	Pydantic AI + Dify API LLM 网关示例
run_server_consumer.py	异步客户端 + 游标轮询示例
run_server_sync_client.py	同步客户端 + `wait_run_sync` 示例
run_server_sse_consumer.py	SSE 流式消费示例
client/_client.py	统一同步/异步 HTTP 客户端实现
adapters/llm/model.py	Dify API LLM 网关到 Pydantic AI 模型的适配层
docs/dify-agent/get-started/index.md	服务器部署与 `.env` 配置参考