1. 为什么要在 Dify FastAPI PyTiDB 里统一模型 Key如果你正在用 Dify 编排一个 TiDB 智能运维 Agent大概率会遇到一个很烦的问题模型 Key 到处散落。Dify 里配一份、FastAPI 服务里写一份、本地调试脚本里再塞一份改一次 Key 要翻三四个文件还容易把测试环境的 Key 带到生产。我这次要做的就是把 Dify 的编排能力、FastAPI 的接口能力、PyTiDB 的数据访问能力串成一条链路同时用 TaoToken 的统一 Key 把模型调用入口收敛到一个地方。先说清楚这套东西是什么、能做什么、适合谁。Dify 是一个开源的大模型应用编排平台你可以把大模型节点、条件分支、工具调用像画流程图一样拖拽串联FastAPI 是一个 Python 的 Web 框架负责把数据库操作暴露成 HTTP 接口PyTiDB 是 TiDB 官方生态的 Python 数据访问层负责把 SQL 请求安全地打到 TiDB 集群。三者组合起来就是一个能用自然语言问 TiDB 集群状态、查慢查询、看会话的运维 Agent。适合谁适合已经在用 Dify 做 Agent、但被多模型 Key 管理折磨的开发者适合想把 TiDB 运维查询接进大模型工作流、又不想把数据库账号密码写死在提示词里的后端同学也适合刚接触 Agent 落地、想找一个能本地跑通的端到端例子的初学者。这篇不讲空概念直接给可复制的配置骨架和一次完整的验证动作目标是让你在本地把通道跑通确认 Key 生效、接口能通、Agent 能查库。核心检索词先摆出来Dify 编排、FastAPI 接口、PyTiDB 连接 TiDB、TaoToken 统一 Key、智能运维 Agent。下面按这个顺序展开。2. TaoToken 前置准备统一 Key 与接入地址在动手写代码之前先把模型调用的入口统一掉。TaoToken 的作用是给你一个统一的 API Key 和统一的接入地址Dify、FastAPI、本地脚本都指向它这样你只需要维护一份 Key。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 接入地址是 https://taotoken.net/api 注意这个 API 地址后面不加 UTM 参数配置里直接写这个就行。你需要先拿到 Key。进入控制台创建 API Key路径是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完复制出来形如sk-xxxxxxxx。这个 Key 就是后面所有配置里唯一要填的凭证。如果你还没决定用哪个模型可以先去模型对话页面试一下 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 确认模型能正常返回再往下走。注意Key 只存在服务端环境变量或本地配置文件里不要提交到 Git也不要写进 Dify 的提示词。Dify 里配置模型供应商时填的是同一个 Key但它是存在 Dify 的数据库里的和你的代码仓库隔离。这里有个容易踩的坑很多人会把 Key 直接写进app.py的常量里结果一提交就泄露。正确做法是走环境变量或者走配置文件代码里只读不写。下面两节分别给config.toml和settings.json两种配置骨架你按自己项目的习惯选一种。3. 可复制配置config.toml 与 settings.json 骨架3.1 config.toml 配置骨架如果你的 FastAPI 服务用 TOML 管理配置可以这样写。这个文件放在项目根目录比如/opt/pytidb-main/config.toml。注意base_url填 TaoToken 的 API 地址api_key从环境变量注入不要硬编码。# /opt/pytidb-main/config.toml [llm] provider openai-compatible base_url https://taotoken.net/api api_key ${TAOTOKEN_API_KEY} model gpt-4o-mini timeout 60 max_retries 2 [tidb] host 192.168.182.31 port 4000 user root password ${TIDB_PASSWORD} database information_schema pool_size 10 max_overflow 20 pool_recycle 3600 [service] host 0.0.0.0 port 8000读取的时候用 Python 的tomllib3.11 自带或者tomli把${TAOTOKEN_API_KEY}这种占位符替换成真实环境变量。这样你的 Key 只存在于 shell 环境里配置文件可以放心提交。import os import tomllib def load_config(path: str config.toml) - dict: with open(path, rb) as f: cfg tomllib.load(f) # 递归替换 ${VAR} 占位符 def resolve(obj): if isinstance(obj, dict): return {k: resolve(v) for k, v in obj.items()} if isinstance(obj, str) and obj.startswith(${) and obj.endswith(}): return os.environ.get(obj[2:-1], ) return obj return resolve(cfg)3.2 settings.json 配置骨架如果你更习惯 JSON或者你的 Dify 自定义工具、外部脚本需要读 JSON可以用这个骨架。字段名和上面 TOML 对齐方便两边切换。{ llm: { provider: openai-compatible, base_url: https://taotoken.net/api, api_key: ${TAOTOKEN_API_KEY}, model: gpt-4o-mini, timeout: 60, max_retries: 2 }, tidb: { host: 192.168.182.31, port: 4000, user: root, password: ${TIDB_PASSWORD}, database: information_schema, pool_size: 10, max_overflow: 20, pool_recycle: 3600 }, service: { host: 0.0.0.0, port: 8000 } }读取逻辑类似用json.load之后做一次占位符替换。两种格式选一种就行不要同时维护两份否则又回到 Key 分散的老问题。提示base_url一定要写https://taotoken.net/api不要带末尾斜杠也不要在后面拼/v1具体路径由 SDK 自己处理。如果你用的是 OpenAI 兼容的客户端它会自动补/chat/completions。3.3 环境变量注入不管用哪种配置文件Key 都从环境变量来。在 systemd 服务里可以这样写# /etc/systemd/system/pytidb-service.service [Unit] DescriptionPyTiDB FastAPI Service for Dify Afternetwork.target [Service] Typesimple Userroot WorkingDirectory/opt/pytidb-main EnvironmentTAOTOKEN_API_KEYsk-你的真实Key EnvironmentTIDB_PASSWORDroot ExecStart/opt/pytidb-main/venv/bin/python3 -m uvicorn app:app --host 0.0.0.0 --port 8000 Restartalways [Install] WantedBymulti-user.target改完执行systemctl daemon-reload systemctl restart pytidb-service.service。这样 Key 只出现在 systemd 配置里代码和 TOML/JSON 都不含明文。4. FastAPI PyTiDB 接入代码与端到端验证4.1 安装依赖先建虚拟环境Python 3.11 比较稳。CentOS 7 的 glibc 偏旧建议用 CentOS 8 或 Rocky Linux能省掉一堆编译报错。python3.11 -m venv /opt/pytidb-main/venv source /opt/pytidb-main/venv/bin/activate pip install --upgrade pip wheel setuptools pip install numpy3,1 --only-binary:all: pip install fastapi uvicorn pymysql sqlalchemy pytidb openaipytidb是 TiDB 官方的 Python 数据访问层sqlalchemypymysql负责连接池和 SQL 执行openai客户端用来调 TaoToken 的统一接口。4.2 app.py 核心代码下面这份app.py把 TiDB 查询接口和模型调用都收进来模型部分读config.toml里的base_url和api_key指向 TaoToken。import os from typing import Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel from sqlalchemy import create_engine, text from sqlalchemy.exc import SQLAlchemyError from openai import OpenAI from config_loader import load_config cfg load_config(config.toml) app FastAPI( titlePyTiDB Service for Dify, descriptionTiDB 集群运维与数据查询接口服务, version0.3.0, ) # ---------- TiDB 连接 ---------- tidb cfg[tidb] db_url ( fmysqlpymysql://{tidb[user]}:{tidb[password]} f{tidb[host]}:{tidb[port]}/{tidb[database]} ) engine create_engine( db_url, pool_recycletidb[pool_recycle], pool_pre_pingTrue, pool_sizetidb[pool_size], max_overflowtidb[max_overflow], ) def run_query(sql_statement: str, params: dict None): try: with engine.connect() as conn: result conn.execute(text(sql_statement), params or {}) if result.returns_rows: return [dict(row._mapping) for row in result] return [] except SQLAlchemyError as e: raise HTTPException(status_code500, detailf数据库查询失败: {str(e)}) # ---------- 模型客户端统一走 TaoToken ---------- llm cfg[llm] client OpenAI( base_urlllm[base_url], api_keyllm[api_key], timeoutllm[timeout], max_retriesllm[max_retries], ) # ---------- 接口 ---------- app.get(/, summary心跳检查) def read_root(): return {status: ok, message: PyTiDB Service is running} app.get(/cluster/nodes, summary获取集群所有节点 IP 及拓扑信息) def get_cluster_nodes(): sql SELECT TYPE, INSTANCE, STATUS, START_TIME FROM INFORMATION_SCHEMA.CLUSTER_INFO; nodes run_query(sql) return {status: success, count: len(nodes), nodes: nodes} app.get(/monitor/slow-queries, summary查询最近的慢查询日志) def get_slow_queries(limit: int 10): sql f SELECT Time, Query_time, Process_time, Wait_time, User, Host, DB, query FROM INFORMATION_SCHEMA.SLOW_QUERY ORDER BY Time DESC LIMIT {limit}; rows run_query(sql) return {status: success, count: len(rows), slow_queries: rows} class AskRequest(BaseModel): question: str app.post(/agent/ask, summary自然语言问 TiDB返回模型回答) def agent_ask(req: AskRequest): # 先取一份集群拓扑作为上下文 nodes run_query( SELECT TYPE, INSTANCE, STATUS FROM INFORMATION_SCHEMA.CLUSTER_INFO; ) context f当前 TiDB 集群节点信息{nodes} resp client.chat.completions.create( modelllm[model], messages[ {role: system, content: 你是 TiDB 运维助手根据给定上下文回答。}, {role: user, content: f{context}\n\n问题{req.question}}, ], ) return {status: success, answer: resp.choices[0].message.content} if __name__ __main__: import uvicorn uvicorn.run(app:app, hostcfg[service][host], portcfg[service][port], reloadTrue)注意config_loader.py就是 3.1 节那段读取逻辑单独放一个文件方便复用。4.3 启动并验证启动服务cd /opt/pytidb-main source venv/bin/activate python app.py先打心跳接口确认服务活着curl http://127.0.0.1:8000/ # {status:ok,message:PyTiDB Service is running}再打集群节点接口确认 PyTiDB 到 TiDB 的链路通curl http://127.0.0.1:8000/cluster/nodes # {status:success,count:6,nodes:[{TYPE:tidb,INSTANCE:192.168.182.31:4000,...}]}最后打 Agent 接口这一步同时验证 TaoToken 的 Key 是否生效、模型是否能返回curl -X POST http://127.0.0.1:8000/agent/ask \ -H Content-Type: application/json \ -d {question:当前集群有几个 TiKV 节点} # {status:success,answer:根据上下文当前集群有 3 个 TiKV 节点...}如果这一步返回了正常回答说明三件事都成了FastAPI 接口通、PyTiDB 查库通、TaoToken 统一 Key 调模型通。这就是一次完整的端到端验证动作。5. 本篇常见错排查5.1 401 Unauthorized 或 invalid api key最常见的原因是TAOTOKEN_API_KEY没注入到进程里。systemd 服务里Environment写了吗改完daemon-reload了吗用systemctl show pytidb-service --propertyEnvironment看一眼实际生效的环境变量。另一个原因是base_url写错必须是https://taotoken.net/api多一个斜杠或者少一个/api都会 404 或 401。5.2 PyTiDB 连接超时或 connection refused先确认 TiDB 的 4000 端口从 FastAPI 所在机器能通telnet 192.168.182.31 4000。如果 FastAPI 和 Dify 在同一台机器、TiDB 在另一台注意防火墙和网段。另外pool_pre_pingTrue能避免连接被 TiDB 主动断开后拿到死连接建议保留。5.3 Dify 自定义工具测试失败Dify 里配 OpenAPI Schema 时servers.url要填 FastAPI 服务的实际地址比如http://192.168.182.153:8000。如果 Dify 跑在 Docker 里而 FastAPI 跑在宿主机127.0.0.1在容器里指向容器自己要用宿主机的局域网 IP 或者 Docker 网桥 IP。测试时先点单个接口的「测试」确认返回 200 再保存。5.4 模型返回乱码或截断检查model字段是不是 TaoToken 支持的模型名。如果返回内容被截断看timeout是不是太短慢查询日志这种大结果集容易超时把timeout调到 120 试试。另外max_retries设 2 能在偶发网络抖动时自动重试。5.5 慢查询接口报权限错误INFORMATION_SCHEMA.SLOW_QUERY需要一定权限如果用的是普通用户可能读不到。临时验证可以用 root生产环境建议单独建一个只读账号只授予PROCESS和SELECT相关权限别直接用 root 跑 Agent。6. 把 Key 收敛之后下一步怎么走走到这里你已经有了一个能本地跑通的 TiDB 智能运维 AgentDify 负责编排和对话FastAPI 暴露接口PyTiDB 连 TiDBTaoToken 统一 Key 管模型调用。Key 只在一处维护改一次全链路生效这是这套架构最实际的价值。如果你接下来要长期跑编码类或 Agent 类任务可以看一下 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它更适合高频调用的场景。如果你还在调接口、排报错先去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 确认 Key 状态再对照接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 检查参数格式。想先验证模型通不通模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 最快。如果你用 Claude Code 做开发Anthropic 兼容入口在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite 。最后留一个我踩过的坑Dify 里配模型供应商时base_url和 FastAPI 里读的base_url要指向同一个 TaoToken 地址否则你会以为 Key 生效了其实是两套配置各走各的。统一入口这件事配一次省心很久。