☰
企业自建 MCP Server 实战:用 Python 打通 ERP 与数据库,TaoToken 统一 Key 接入
2026/10/11 13:52:56 网站建设 项目流程

1. 企业内网为什么需要自建 MCP Server

很多团队在做企业 AI Agent 落地时,卡住的地方往往不是模型本身,而是"接口不通"。ERP 里躺着订单、库存、客户数据,数据库里有实时业务表,但这些系统各自有独立的鉴权方式、数据格式和调用约定。Agent 想查一条订单状态,要么让后端临时写个接口,要么在 Agent 里硬编码一段 SQL 和连接串。

当 Agent 只有一个的时候,硬编码还能忍。但当问答 Agent、报表 Agent、审批 Agent 同时需要访问 ERP 和数据库时,每个 Agent 都要重复对接一遍,后端接口一改,所有 Agent 跟着改。这种维护成本会随着 Agent 数量呈指数上升。

MCP(Model Context Protocol)解决的正是这个问题。它把"工具接入"从定制化开发变成标准化配置:你写一个 MCP Server 把 ERP 查询和数据库读写封装成标准工具,所有支持 MCP 的 Agent 客户端都能即插即用地调用,不需要各自重复对接。MCP Server 本质上是一个暴露工具(Tool)和资源(Resource)的服务端进程,通过 stdio 或 SSE 与客户端通信,工具就是 Agent 可以主动调用的一次性函数,比如query_orders(status)。

这篇文章面向后端工程师和 AI 应用开发者,聚焦企业内网自建 MCP Server 的落地路径。我会用 Python 3.12 + MCP Python SDK 从零搭一个能跑的 Server,把 ERP 查询和数据库读写封装成可调用工具,然后说明怎么把服务端 endpoint 与鉴权配置改到 TaoToken 统一 Key/API 通道。全文给出可复制的目录结构、SDK 注册代码、ERP/数据库连接参数模板,以及用 curl 与 MCP 客户端各验证一次工具调用的具体动作。适合已经在跑至少一个 Agent、准备把内部系统接进来的团队。

2. TaoToken 统一 Key 接入前置准备

在动手写 Server 之前,先把鉴权和通道这层理清楚。企业内网自建 MCP Server 有一个绕不开的问题:Server 本身要调用模型能力做意图理解或结果润色时,每个 Agent、每个服务都各自持有一份模型 Key,管理起来非常乱。TaoToken 在这里的作用是提供统一的 Key/API 通道,让 MCP Server 和 Agent 客户端共用一套鉴权入口,而不是每个组件单独配 Key。

你需要先拿到一个 TaoToken 的 API Key。登录官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 注册后,进入控制台 https://taotoken.net/console 创建 API Key。创建完成后在 API Keys 页面 https://taotoken.net/api-keys 可以看到完整的 Key 字符串,格式通常是sk-开头的一串字符。这个 Key 就是后面 MCP Server 和客户端统一使用的凭证。

TaoToken 的 API 基地址是 https://taotoken.net/api,注意这个地址不带任何查询参数。所有走 OpenAI 兼容协议的请求都发到这个 Base URL,路径拼接/v1/chat/completions即可。如果你用的是 Claude Code 这类 Anthropic 协议客户端,接入文档在 https://taotoken.net/doc 有详细说明,Claude Code 专用接入页在 https://taotoken.net/ClaudeCodeAnthropic。

环境变量建议这样组织,把 Key 和 Base URL 都收敛到一处,避免散落在各个脚本里:

# ~/.mcp_env 或 systemd EnvironmentFile export TAOTOKEN_API_KEY="sk-你的实际Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export ERP_DB_PATH="/data/erp.db" export ERP_TOKEN="内网工具调用令牌"

这里要区分两个概念:TAOTOKEN_API_KEY是访问 TaoToken 模型通道用的,ERP_TOKEN是 MCP Server 内部校验工具调用权限用的,两者不要混用。前者管"能不能调模型",后者管"能不能查 ERP"。生产环境里这两个值都应该走密钥管理服务,不要明文写在代码或配置文件里提交到仓库。

Python 环境方面,推荐 3.11 以上,本文用 3.12。MCP Python SDK 要求mcp>=1.2.0。先建隔离环境再装依赖:

python3.12 -m venv .venv source .venv/bin/activate pip install "mcp[cli]>=1.2.0" python -c "import mcp; print(mcp.__version__)"

预期输出1.2.0或更高。如果这一步报ModuleNotFoundError,基本是虚拟环境没激活或者 pip 装到了系统 Python 里,用which python确认一下路径指向.venv/bin/python。

目录结构建议这样组织,把 Server、配置、数据分层放,方便后面加工具和排障:

mcp-erp-bridge/ ├── .venv/ ├── erp_bridge.py # MCP Server 主文件 ├── config/ │ └── settings.toml # 连接参数与鉴权配置 ├── data/ │ └── erp.db # 演示用 SQLite,生产换 PostgreSQL └── client_test.py # MCP 客户端验证脚本

3. 可复制的 MCP Server 配置与工具注册代码

这一节是全文的核心,给出可以直接复制运行的配置片段和 SDK 注册代码。先看配置文件,用 TOML 管理连接参数,路径和字段名保持和代码一致:

# config/settings.toml [taotoken] base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" model_id = "claude-3-5-sonnet" [erp] db_path = "/data/erp.db" auth_token_env = "ERP_TOKEN" query_timeout = 5 [server] name = "erp-bridge" transport = "stdio"

注意model_id这个字段,它是 MCP Server 在需要调用模型做结果润色时使用的模型标识。TaoToken 统一 Key 通道下,Base URL、API Key、Model ID 这三件套要配全,缺一个都会在调用时报错。如果你用的是 Cline 或 CC Switch 这类客户端,它们的 MCP 配置里同样需要这三项,格式如下:

{ "mcpServers": { "erp-bridge": { "command": "python", "args": ["/opt/mcp/erp_bridge.py"], "env": { "TAOTOKEN_API_KEY": "sk-你的实际Key", "TAOTOKEN_BASE_URL": "https://taotoken.net/api", "ERP_TOKEN": "内网工具调用令牌" } } } }

接下来是 Server 主文件。用 FastMCP 高层封装,装饰器注册工具,代码量很少:

# erp_bridge.py # 依赖:Python 3.12 + mcp>=1.2.0 # 运行:python erp_bridge.py 或 mcp run erp_bridge.py import os import sqlite3 import tomllib from mcp.server.fastmcp import FastMCP # 读取配置 with open("config/settings.toml", "rb") as f: cfg = tomllib.load(f) EXPECTED_TOKEN = os.environ.get(cfg["erp"]["auth_token_env"], "") DB_PATH = cfg["erp"]["db_path"] mcp = FastMCP(cfg["server"]["name"]) def _auth(token: str) -> bool: """校验工具调用令牌,生产环境应换成密钥管理服务""" return token == EXPECTED_TOKEN and EXPECTED_TOKEN != "" @mcp.tool() def query_orders(status: str, token: str = "") -> list: """按状态查询 ERP 中的订单,返回订单列表""" if not _auth(token): raise PermissionError("invalid token") conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute( "SELECT id, customer, amount FROM orders WHERE status=?", (status,), ) rows = cur.fetchall() conn.close() return [{"id": r[0], "customer": r[1], "amount": r[2]} for r in rows] @mcp.tool() def update_order_status(order_id: int, new_status: str, token: str = "") -> dict: """更新指定订单的状态,返回受影响行数""" if not _auth(token): raise PermissionError("invalid token") conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute( "UPDATE orders SET status=? WHERE id=?", (new_status, order_id), ) conn.commit() affected = cur.rowcount conn.close() return {"order_id": order_id, "affected": affected} if __name__ == "__main__": mcp.run(transport=cfg["server"]["transport"])

这里有两个工具:query_orders负责读,update_order_status负责写。每个工具都带token参数做鉴权,这是内网自建 Server 的基本要求,避免工具被越权调用。_auth函数里加了EXPECTED_TOKEN != ""的判断,防止环境变量没配时令牌为空导致校验被绕过。

如果你需要 Server 在返回结果前调用模型做润色,可以加一个走 TaoToken 通道的辅助函数:

import httpx def polish_with_model(text: str) -> str: """通过 TaoToken 统一通道调用模型润色文本""" base = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") key = os.environ.get("TAOTOKEN_API_KEY", "") resp = httpx.post( f"{base}/v1/chat/completions", headers={"Authorization": f"Bearer {key}"}, json={ "model": cfg["taotoken"]["model_id"], "messages": [{"role": "user", "content": f"请简洁润色:{text}"}], }, timeout=10, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]

注意 Base URL 拼接的是/v1/chat/completions,不要重复写/api。TaoToken 的 Base URL 已经包含了/api前缀,再拼一次会变成/api/api/v1/...导致 404。

4. 验证请求与成功结果

配置写完,必须验证两件事:Server 能不能正常启动并暴露工具,以及工具调用能不能返回预期结果。分两步走,先用 curl 验证 TaoToken 通道本身通不通,再用 MCP 客户端验证工具调用。

先验证 TaoToken 通道。这一步确认你的 Key 和 Base URL 是对的:

curl -s https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "claude-3-5-sonnet", "messages": [{"role": "user", "content": "回复 OK 两个字母"}] }'

成功的话返回 JSON 里choices[0].message.content会是OK。如果返回 401,说明 Key 不对或没带上;如果返回 404,检查 Base URL 是不是多拼了路径。

接着验证 MCP Server 的工具调用。写一个客户端脚本,用 stdio 连上 Server 并调用query_orders:

# client_test.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params = StdioServerParameters( command="python", args=["erp_bridge.py"], env={ "TAOTOKEN_API_KEY": "sk-你的实际Key", "TAOTOKEN_BASE_URL": "https://taotoken.net/api", "ERP_TOKEN": "内网工具调用令牌", }, ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await session.list_tools() print("可用工具:", [t.name for t in tools.tools]) result = await session.call_tool( "query_orders", {"status": "paid", "token": "内网工具调用令牌"}, ) print("查询结果:", result.content) asyncio.run(main())

运行python client_test.py,预期输出类似:

可用工具: ['query_orders', 'update_order_status'] 查询结果: [{'id': 1001, 'customer': '某客户', 'amount': 2999.0}]

看到工具列表和查询结果,说明整条链路通了:客户端通过 stdio 连上 Server,Server 校验令牌后查了 SQLite,把结果按 MCP 协议返回。如果list_tools返回空列表,说明装饰器没生效或者 Server 启动时抛了异常,把mcp.run换成直接调用工具函数先排查业务逻辑。

再验证一次写操作,确认update_order_status能改数据:

result = await session.call_tool( "update_order_status", {"order_id": 1001, "new_status": "shipped", "token": "内网工具调用令牌"}, ) print("更新结果:", result.content)

预期返回{'order_id': 1001, 'affected': 1}。affected为 0 说明订单 ID 不存在,为 1 说明更新成功。这两步都过了,Server 就算真正可用了。

5. 常见报错排查

实际部署时踩的坑基本集中在几类报错上,逐个对照排查。

401 Unauthorized:出现在 curl 调 TaoToken 或 Server 调模型时。先确认TAOTOKEN_API_KEY环境变量在当前 shell 里真的存在,用echo $TAOTOKEN_API_KEY看输出。如果为空,说明source ~/.mcp_env没执行或者变量名拼错。如果 Key 存在但仍 401,去 API Keys 页面 https://taotoken.net/api-keys 确认 Key 没被删除或过期。注意 Header 格式必须是Authorization: Bearer sk-xxx,少个空格也会 401。

local proxy failed / connection refused:MCP 客户端连不上 Server。90% 是 Python 路径问题。StdioServerParameters里的command如果写python,客户端会用它自己的 PATH 去找,可能找到系统 Python 而不是虚拟环境里的。改成绝对路径/opt/mcp/.venv/bin/python最稳。可以在 Server 脚本首行加import sys; print(sys.executable, file=sys.stderr)确认实际用的是哪个解释器。

reading choices 报错 / KeyError: 'choices':调模型返回的 JSON 里没有choices字段。通常是 Base URL 拼错导致请求打到了错误端点,或者模型 ID 写错。检查TAOTOKEN_BASE_URL是不是https://taotoken.net/api,请求路径是不是/v1/chat/completions。如果返回体里有error字段,先打印出来看具体信息。

OAuth 相关报错:如果你用的是 Claude Code 或 Codex 这类带 OAuth 流程的客户端,报 OAuth 失败通常是客户端配置里的鉴权方式没切到 API Key 模式。Claude Code 接入参考 https://taotoken.net/ClaudeCodeAnthropic,Codex 的auth.json里要确保OPENAI_API_KEY字段填的是 TaoToken 的 Key,OPENAI_BASE_URL填https://taotoken.net/api。三件套 Base URL、Key、Model ID 缺一不可。

database is locked:多 Agent 并发调用 SQLite 时出现。SQLite 单写连接不支持高并发写,生产环境要么换成 PostgreSQL,要么在 Server 里加连接池和写锁。演示阶段可以给sqlite3.connect加timeout=10参数缓解。

工具返回中文乱码:stdio 传输默认 UTF-8,但部分旧系统 locale 不是 UTF-8。在启动命令前加env LANG=C.UTF-8强制编码,或者在StdioServerParameters的env里显式设置LANG。

list_tools 返回空:装饰器没生效。检查@mcp.tool()是否写在函数正上方,函数是否有类型注解。FastMCP 依赖类型注解生成工具 schema,缺注解会导致工具注册失败但不报错。

6. 统一 Key 通道下的接入与后续扩展

把 Server 跑通只是第一步,真正落地要考虑的是怎么让多个 Agent 复用同一套工具,以及鉴权怎么收敛。TaoToken 统一 Key 通道在这里的价值是:MCP Server 和所有 Agent 客户端共用一套模型鉴权入口,不用每个组件单独配 Key,也不用担心 Key 散落在各个配置文件里。

多 Agent 复用的做法很简单,同一个 Server 可以被多个客户端同时挂载。问答 Agent 调query_orders查订单,报表 Agent 调同一个工具做统计,审批 Agent 调update_order_status改状态,它们连的是同一个 Server 进程,工具只写一次。MCP 会话是隔离的,不会串数据,但要注意 SQLite 的并发写限制。

如果你需要跨机器调用,把transport从stdio改成sse,Server 会监听一个 HTTP 端口。这时候必须配 HTTPS 和鉴权,因为 SSE 模式会暴露网络端口。内网部署建议加 mTLS,确保只有受信任的客户端能连。stdio 模式最安全,不需要开端口,适合单机本地部署。

后续扩展方向有两个。一是加 Resource,让 Agent 不只是"调用"工具,还能"感知"业务状态,比如订阅订单表的变化。二是加 Prompt 模板,把常用的查询模式固化成提示,减少 Agent 每次重新构造参数的负担。这两个原语和 Tool 一样,都是 MCP 协议标准的一部分,FastMCP 都有对应的装饰器。

需要长期跑编码类 Agent 或做多 Agent 编排的团队,可以了解下 Coding Plan https://taotoken.net/coding-plan,它针对持续编码场景做了通道优化。想先验证模型对话效果的,直接去模型对话页 https://taotoken.net/chat 试一下。接入过程中遇到协议细节问题,文档 https://taotoken.net/doc 里有完整的参数说明和示例。

最后提醒一点:MCP Server 持有业务系统凭证,必须放在内网且与 Agent 同可用区。对外暴露 SSE 时务必启用 mTLS,涉及核心经营数据的场景优先用 stdio 本地模式,从根本上杜绝数据出域风险。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询