尧图网站建设 
1. 内网数据服务器上部署 MCP 服务器给 Workbuddy 调用的真实场景很多团队的数据服务器压根不对外Oracle、MySQL 全在内网跑着但业务方又想用 Workbuddy 这类 AI 工具直接问「上个月华东区退货率多少」然后拿到报表。直接开数据库端口给 AI 工具安全团队第一个不同意。这时候 MCP 服务器就是那个「中间人」——它跑在内网数据服务器上只暴露白名单里的表和列Workbuddy 通过 MCP 协议调用它数据不出内网权限还能精确到列级别。我这次要落地的场景很具体一台已经跑着 Oracle 的数据服务器上面有空余资源不新购机器用 Python 部署一个 MCP 服务器对外只开一个 SSE 端口。Workbuddy 企业版通过 MCP 配置连上来用户用自然语言提问MCP 服务器把请求翻译成受限 SQL 查库返回 Markdown 表格。关键点在于——MCP 的 endpoint 要统一改到 TaoToken 通道这样模型侧的调用走同一个入口密钥管理、用量统计、模型切换都在一个地方收口不用每个工具单独配一遍。适合谁看如果你手上有内网数据库、团队在用 Workbuddy 或类似 AI 助手、又不想把库直接暴露出去这套流程可以直接复现。下面从环境准备到连通性验证一步步来。2. TaoToken 前置准备MCP 服务端与 Workbuddy 的通道统一在动手写 MCP 服务器代码之前先把 TaoToken 这条通道理清楚。MCP 服务器本身负责「查数据」但 Workbuddy 在理解用户意图、生成 SQL 草稿、把结果转成自然语言这几步需要调用大模型。如果每个环节各自配一套模型密钥后面排查问题会非常痛苦。把模型调用统一指向 TaoTokenMCP 服务端和 Workbuddy 侧共用同一个 Base URL 和 Key链路就清晰了。TaoToken 的 API 入口是https://taotoken.net/api兼容 OpenAI 风格的接口格式。你需要先在控制台创建一个 API Key这个 Key 后面会同时出现在 MCP 服务端的环境变量和 Workbuddy 的 MCP 配置里。模型 ID 建议选一个支持工具调用function calling的因为 MCP 的 tool 定义需要模型能理解并生成结构化参数。具体操作路径打开https://taotoken.net/console登录后进入 API Keys 页面点创建复制出来的 Key 形如sk-xxxxxxxx。这个 Key 只显示一次先存到密码管理器里。然后到模型对话页面确认一下你要用的模型 ID 是否可用比如claude-sonnet-4-20250514这类支持 tool use 的模型。如果你打算长期跑编码类 Agent 任务可以顺带看一下 Coding Plan 的额度说明避免后面调用量上来触发限流。这里有个容易踩的坑MCP 服务端和 Workbuddy 侧用的 Key 可以是同一个但环境变量名要区分开。MCP 服务端我习惯用TAOTOKEN_API_KEYWorkbuddy 的 MCP 配置里直接写Authorization头。两边都指向https://taotoken.net/api不要一个写带/v1一个不写后面 401 报错十有八九是路径不一致导致的。另外MCP 服务器部署在内网它访问 TaoToken 需要出网。如果你的数据服务器完全隔离需要让网络组开一条到taotoken.net的出站策略只放 443 端口。这一步提前和网络同事沟通好不然代码写完了连不通模型白忙活。3. 可复制配置MCP 服务端代码与 Workbuddy 侧参数先给 MCP 服务端的完整配置。我用 Python 写依赖mcp、oracledb、starlette、uvicorn。在数据服务器上建一个目录/opt/mcp-oracle把下面的代码存成server.py。注意环境变量部分数据库账号密码和 TaoToken 的 Key 都从环境变量读不要硬编码。# codingutf-8 import asyncio import logging import os import oracledb from typing import Any, Dict, List from mcp.server import Server from mcp.types import Tool, TextContent from starlette.applications import Starlette from starlette.routing import Route from starlette.requests import Request from starlette.responses import Response from mcp.server.sse import SseServerTransport import uvicorn logging.basicConfig(levellogging.INFO) logger logging.getLogger(MCP-Oracle-Server) # 数据库配置从环境变量读取 host os.environ.get(DB_HOST, 127.0.0.1) port int(os.environ.get(DB_PORT, 1521)) oracleuser os.environ.get(DB_USER, ) oraclepassword os.environ.get(DB_PASSWORD, ) service_name os.environ.get(DB_SERVICE, ) # TaoToken 通道配置 TAOTOKEN_BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) TAOTOKEN_MODEL os.environ.get(TAOTOKEN_MODEL, claude-sonnet-4-20250514) if not all([oracleuser, oraclepassword, service_name]): logger.warning(缺少数据库环境变量服务可能无法连接) DB_CONFIG { host: host, port: port, service_name: service_name, user: oracleuser, password: oraclepassword, } app Server(oracle-data-connector) # 白名单表名 - 允许查询的列 ALLOWED_TABLES { SALES.ORDERS: [ORDER_ID, CUSTOMER_ID, AMOUNT, STATUS, CREATED_AT], SALES.CUSTOMERS: [CUSTOMER_ID, NAME, REGION, LEVEL], } pool None def init_oracle_client(): try: client_dir os.environ.get(ORACLE_CLIENT_DIR, ) if client_dir: oracledb.init_oracle_client(lib_dirclient_dir) logger.info(Oracle Thick 模式初始化完成) except Exception as e: logger.error(fOracle 客户端初始化失败: {e}) raise def init_connection_pool(): global pool if pool is None: dsn f{DB_CONFIG[host]}:{DB_CONFIG[port]}/{DB_CONFIG[service_name]} pool oracledb.create_pool( userDB_CONFIG[user], passwordDB_CONFIG[password], dsndsn, min1, max4, increment1, getmodeoracledb.POOL_GETMODE_WAIT, ) logger.info(Oracle 连接池初始化完成) def get_db_connection(): if pool is None: init_connection_pool() return pool.acquire() app.list_tools() async def list_tools() - List[Tool]: return [ Tool( namelist_allowed_tables, description列出当前 MCP 服务允许查询的所有表名, inputSchema{type: object, properties: {}, required: []}, ), Tool( namequery_oracle_table, description查询指定 Oracle 表的数据仅支持白名单中的表和列, inputSchema{ type: object, properties: { table_name: {type: string, description: 要查询的表名必须大写}, columns: { type: array, items: {type: string}, description: 要查询的列名列表为空则返回所有允许列, }, limit: {type: integer, description: 返回行数上限默认 10最大 100, default: 10}, filter_column: {type: string, description: 可选过滤列}, filter_value: {type: string, description: 可选过滤值}, }, required: [table_name], }, ), ] app.call_tool() async def call_tool(name: str, arguments: Dict[str, Any]) - List[TextContent]: try: if name list_allowed_tables: tables list(ALLOWED_TABLES.keys()) return [TextContent(typetext, textf允许查询的表: {, .join(tables)})] elif name query_oracle_table: table_name arguments.get(table_name, ).upper() requested_columns arguments.get(columns, []) limit min(arguments.get(limit, 10), 100) filter_col arguments.get(filter_column) filter_val arguments.get(filter_value) if table_name not in ALLOWED_TABLES: return [TextContent(typetext, textf错误: 表 {table_name} 不在白名单中)] allowed_cols ALLOWED_TABLES[table_name] if not requested_columns or * in requested_columns: select_cols allowed_cols else: invalid_cols [c for c in requested_columns if c.upper() not in allowed_cols] if invalid_cols: return [TextContent(typetext, textf错误: 列 {invalid_cols} 不允许查询)] select_cols [c.upper() for c in requested_columns] col_str , .join(select_cols) params {} base_sql fSELECT {col_str} FROM {table_name} if filter_col and filter_val is not None: if filter_col.upper() not in allowed_cols: return [TextContent(typetext, textf错误: 过滤列 {filter_col} 不允许)] base_sql f WHERE {filter_col.upper()} :bind_val params[:bind_val] filter_val sql fSELECT * FROM ({base_sql}) WHERE ROWNUM {limit} conn get_db_connection() cursor conn.cursor() try: cursor.execute(sql, params) columns [desc[0] for desc in cursor.description] rows cursor.fetchall() result_text f查询结果 {table_name}:\n result_text | | .join(columns) |\n result_text | |.join([--- for _ in columns]) |\n for row in rows: formatted [str(item) if item is not None else NULL for item in row] result_text | | .join(formatted) |\n return [TextContent(typetext, textresult_text)] finally: cursor.close() conn.close() else: return [TextContent(typetext, textf未知工具: {name})] except Exception as e: logger.error(f工具执行错误 {name}: {e}) return [TextContent(typetext, textf错误: {str(e)})] def main(): init_oracle_client() try: init_connection_pool() except Exception: logger.error(数据库连接池初始化失败MCP 服务不启动) return sse SseServerTransport(/messages/) async def handle_sse(request: Request): async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) async def handle_messages(request: Request): await sse.handle_post_message(request.scope, request.receive, request._send) return Response(status_code200) starlette_app Starlette( debugTrue, routes[ Route(/sse, endpointhandle_sse), Route(/messages/, endpointhandle_messages, methods[POST]), Route(/messages, endpointhandle_messages, methods[POST]), ], ) logger.info(启动 Oracle MCP Server端口 8080) uvicorn.run(starlette_app, host0.0.0.0, port8080) if __name__ __main__: main()环境变量文件/opt/mcp-oracle/.env这样写DB_HOST127.0.0.1 DB_PORT1521 DB_USERmcp_readonly DB_PASSWORD你的数据库密码 DB_SERVICEORCLPDB1 ORACLE_CLIENT_DIR/opt/oracle/instantclient_21_12 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的TaoToken密钥 TAOTOKEN_MODELclaude-sonnet-4-20250514Workbuddy 侧的 MCP 配置在设置里找到 MCP Servers粘贴下面这段 JSON。注意url里的 IP 换成你数据服务器的内网地址端口和代码里一致{ mcpServers: { Oracle-Connector: { transport: sse, url: http://10.0.1.50:8080/sse, disabled: false, headers: { Authorization: Bearer sk-你的TaoToken密钥 } } } }如果你用的是 Cline 或 Claude Code 这类支持 MCP 的客户端配置结构类似把transport和url对应填好即可。Codex 用户如果走auth.json把base_url指向https://taotoken.net/apiapi_key填同一个 Key。三件套Base URL Key Model ID在 MCP 服务端和客户端两侧保持一致后面排查会省很多事。4. 验证请求一次完整的连通性检查与成功结果配置写完了别急着在 Workbuddy 里问业务问题先做一次最小连通性验证。分三步MCP 服务端自检、SSE 端点探测、Workbuddy 工具调用。第一步在数据服务器上启动服务cd /opt/mcp-oracle set -a source .env set a python server.py看到启动 Oracle MCP Server端口 8080和Oracle 连接池初始化完成就说明服务端起来了。如果卡在连接池初始化检查数据库账号密码和 service_name。第二步在另一台能访问数据服务器的机器上探测 SSE 端点curl -N -H Accept: text/event-stream http://10.0.1.50:8080/sse正常会返回类似event: endpoint和data: /messages/?session_idxxx的内容连接保持不关闭。如果 curl 直接报Connection refused检查防火墙和 uvicorn 监听地址如果返回 404检查路由路径是不是/sse。第三步打开 Workbuddy在对话里明确指定使用 Oracle-Connector 这个 MCP。比如输入「用 Oracle-Connector 查询 SALES.ORDERS 表的前 5 行列出 ORDER_ID 和 AMOUNT」。Workbuddy 会先调用list_allowed_tables确认表在不在白名单再调用query_oracle_table带参数查询。成功的话你会看到类似下面的返回查询结果 SALES.ORDERS: | ORDER_ID | AMOUNT | |---|---| | 1001 | 299.00 | | 1002 | 1580.50 | | 1003 | 76.20 | | 1004 | 4300.00 | | 1005 | 215.75 |到这一步整条链路就通了Workbuddy 理解意图 → 通过 TaoToken 通道调用模型生成工具参数 → MCP 服务器执行受限 SQL → 返回 Markdown 表格 → 模型整理成自然语言回复。你可以再试一个带过滤条件的「查 SALES.CUSTOMERS 里 REGION 是 East 的客户只要 NAME 和 LEVEL 两列」验证白名单列过滤和 bind 参数是否生效。实测下来从启动服务到 Workbuddy 拿到第一份报表顺利的话十分钟内能跑通。真正花时间的是数据库账号权限配置和网络策略开通代码本身不复杂。5. 本篇常见错误排查401、local proxy failed 与 reading choices 报错部署过程中最容易撞上的几个报错我按出现频率排一下。401 UnauthorizedWorkbuddy 侧调用 MCP 时返回 401九成是Authorization头没带对。检查 JSON 配置里headers的 Key 是不是Authorization值是不是Bearer sk-xxxBearer 和 Key 之间有一个空格。另一个可能是 TaoToken 的 Key 过期或被删了去控制台重新生成一个换上。如果 MCP 服务端日志里看到模型调用 401检查.env里TAOTOKEN_API_KEY有没有被 shell 转义吃掉用echo $TAOTOKEN_API_KEY确认一下。local proxy failed这个报错通常出现在 Workbuddy 尝试连接 MCP 的 SSE 端点时。原因一般是网络不通——Workbuddy 所在机器访问不到数据服务器的 8080 端口。先在 Workbuddy 机器上telnet 10.0.1.50 8080测一下不通就找网络组开策略。还有一种情况是 MCP 服务端绑定了127.0.0.1而不是0.0.0.0外部访问不了检查uvicorn.run的 host 参数。reading choices 报错这个一般出现在模型返回结果解析阶段提示读取choices字段失败。根因通常是 TaoToken 通道返回的响应格式和客户端预期不一致。检查TAOTOKEN_BASE_URL是不是写成了https://taotoken.net/api/v1而客户端又自动拼了/v1导致路径变成/api/v1/v1/chat/completions。统一用https://taotoken.net/api让客户端自己拼版本路径。另外确认模型 ID 拼写正确不存在的模型 ID 有时会返回非标准错误体客户端解析时就报 reading choices。OAuth 相关报错如果你用的客户端走 OAuth 流程接 MCP报OAuth token invalid或redirect_uri mismatch先确认是不是把 API Key 认证和 OAuth 认证混用了。MCP 的 SSE 传输用 Bearer Token 就够了不需要走 OAuth。如果客户端强制 OAuth检查回调地址有没有在 TaoToken 控制台登记。数据库连接超时MCP 服务端日志报DPY-6005或连接池获取超时检查数据库监听是否正常、service_name是否写成了 SID。Oracle 的 service_name 和 SID 是两个东西用lsnrctl status确认一下。连接池max4在高并发下可能不够但内网小团队用足够了真不够再调。排查顺序建议先看 MCP 服务端日志再看 Workbuddy 侧报错最后用 curl 直接打 SSE 端点。日志里logger.error打出来的信息比客户端报错详细得多养成先看服务端日志的习惯。6. 把 MCP endpoint 统一到 TaoToken 后的日常维护通道统一之后日常维护就简单了。所有模型调用都经过 TaoToken你在控制台能看到每个 Key 的调用量和费用MCP 服务端和 Workbuddy 侧共用同一个 Key账单不会分散。要换模型改.env里的TAOTOKEN_MODEL重启服务就行Workbuddy 侧不用动。要加新的表白名单改ALLOWED_TABLES字典重启权限控制始终在 MCP 服务端这一层。API Key 的轮换也方便在控制台新建一个 Key把.env和 Workbuddy 配置里的旧 Key 替换掉重启 MCP 服务旧 Key 在控制台禁用。整个过程不影响数据库也不需要在多个工具之间同步修改。如果你后面要接更多数据源比如 MySQL 或 PostgreSQL可以再起一个 MCP 服务实例监听不同端口Workbuddy 侧加一个mcpServers条目就行。每个实例独立白名单、独立连接池互不干扰。模型通道还是同一个 TaoToken管理成本不随数据源数量线性增长。最后提醒一句MCP 服务端的白名单一定要认真配别图省事用SELECT *放开所有列。AI 工具再智能权限边界还是得靠代码守住。数据库账号用只读账号只授白名单表的 SELECT 权限双保险。 数据服务器+MCP服务器部署MCP供Workbuddy调用：把endpoint改到TaoToken的完整配置
1. 内网数据服务器上部署 MCP 服务器给 Workbuddy 调用的真实场景很多团队的数据服务器压根不对外Oracle、MySQL 全在内网跑着但业务方又想用 Workbuddy 这类 AI 工具直接问「上个月华东区退货率多少」然后拿到报表。直接开数据库端口给 AI 工具安全团队第一个不同意。这时候 MCP 服务器就是那个「中间人」——它跑在内网数据服务器上只暴露白名单里的表和列Workbuddy 通过 MCP 协议调用它数据不出内网权限还能精确到列级别。我这次要落地的场景很具体一台已经跑着 Oracle 的数据服务器上面有空余资源不新购机器用 Python 部署一个 MCP 服务器对外只开一个 SSE 端口。Workbuddy 企业版通过 MCP 配置连上来用户用自然语言提问MCP 服务器把请求翻译成受限 SQL 查库返回 Markdown 表格。关键点在于——MCP 的 endpoint 要统一改到 TaoToken 通道这样模型侧的调用走同一个入口密钥管理、用量统计、模型切换都在一个地方收口不用每个工具单独配一遍。适合谁看如果你手上有内网数据库、团队在用 Workbuddy 或类似 AI 助手、又不想把库直接暴露出去这套流程可以直接复现。下面从环境准备到连通性验证一步步来。2. TaoToken 前置准备MCP 服务端与 Workbuddy 的通道统一在动手写 MCP 服务器代码之前先把 TaoToken 这条通道理清楚。MCP 服务器本身负责「查数据」但 Workbuddy 在理解用户意图、生成 SQL 草稿、把结果转成自然语言这几步需要调用大模型。如果每个环节各自配一套模型密钥后面排查问题会非常痛苦。把模型调用统一指向 TaoTokenMCP 服务端和 Workbuddy 侧共用同一个 Base URL 和 Key链路就清晰了。TaoToken 的 API 入口是https://taotoken.net/api兼容 OpenAI 风格的接口格式。你需要先在控制台创建一个 API Key这个 Key 后面会同时出现在 MCP 服务端的环境变量和 Workbuddy 的 MCP 配置里。模型 ID 建议选一个支持工具调用function calling的因为 MCP 的 tool 定义需要模型能理解并生成结构化参数。具体操作路径打开https://taotoken.net/console登录后进入 API Keys 页面点创建复制出来的 Key 形如sk-xxxxxxxx。这个 Key 只显示一次先存到密码管理器里。然后到模型对话页面确认一下你要用的模型 ID 是否可用比如claude-sonnet-4-20250514这类支持 tool use 的模型。如果你打算长期跑编码类 Agent 任务可以顺带看一下 Coding Plan 的额度说明避免后面调用量上来触发限流。这里有个容易踩的坑MCP 服务端和 Workbuddy 侧用的 Key 可以是同一个但环境变量名要区分开。MCP 服务端我习惯用TAOTOKEN_API_KEYWorkbuddy 的 MCP 配置里直接写Authorization头。两边都指向https://taotoken.net/api不要一个写带/v1一个不写后面 401 报错十有八九是路径不一致导致的。另外MCP 服务器部署在内网它访问 TaoToken 需要出网。如果你的数据服务器完全隔离需要让网络组开一条到taotoken.net的出站策略只放 443 端口。这一步提前和网络同事沟通好不然代码写完了连不通模型白忙活。3. 可复制配置MCP 服务端代码与 Workbuddy 侧参数先给 MCP 服务端的完整配置。我用 Python 写依赖mcp、oracledb、starlette、uvicorn。在数据服务器上建一个目录/opt/mcp-oracle把下面的代码存成server.py。注意环境变量部分数据库账号密码和 TaoToken 的 Key 都从环境变量读不要硬编码。# codingutf-8 import asyncio import logging import os import oracledb from typing import Any, Dict, List from mcp.server import Server from mcp.types import Tool, TextContent from starlette.applications import Starlette from starlette.routing import Route from starlette.requests import Request from starlette.responses import Response from mcp.server.sse import SseServerTransport import uvicorn logging.basicConfig(levellogging.INFO) logger logging.getLogger(MCP-Oracle-Server) # 数据库配置从环境变量读取 host os.environ.get(DB_HOST, 127.0.0.1) port int(os.environ.get(DB_PORT, 1521)) oracleuser os.environ.get(DB_USER, ) oraclepassword os.environ.get(DB_PASSWORD, ) service_name os.environ.get(DB_SERVICE, ) # TaoToken 通道配置 TAOTOKEN_BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) TAOTOKEN_MODEL os.environ.get(TAOTOKEN_MODEL, claude-sonnet-4-20250514) if not all([oracleuser, oraclepassword, service_name]): logger.warning(缺少数据库环境变量服务可能无法连接) DB_CONFIG { host: host, port: port, service_name: service_name, user: oracleuser, password: oraclepassword, } app Server(oracle-data-connector) # 白名单表名 - 允许查询的列 ALLOWED_TABLES { SALES.ORDERS: [ORDER_ID, CUSTOMER_ID, AMOUNT, STATUS, CREATED_AT], SALES.CUSTOMERS: [CUSTOMER_ID, NAME, REGION, LEVEL], } pool None def init_oracle_client(): try: client_dir os.environ.get(ORACLE_CLIENT_DIR, ) if client_dir: oracledb.init_oracle_client(lib_dirclient_dir) logger.info(Oracle Thick 模式初始化完成) except Exception as e: logger.error(fOracle 客户端初始化失败: {e}) raise def init_connection_pool(): global pool if pool is None: dsn f{DB_CONFIG[host]}:{DB_CONFIG[port]}/{DB_CONFIG[service_name]} pool oracledb.create_pool( userDB_CONFIG[user], passwordDB_CONFIG[password], dsndsn, min1, max4, increment1, getmodeoracledb.POOL_GETMODE_WAIT, ) logger.info(Oracle 连接池初始化完成) def get_db_connection(): if pool is None: init_connection_pool() return pool.acquire() app.list_tools() async def list_tools() - List[Tool]: return [ Tool( namelist_allowed_tables, description列出当前 MCP 服务允许查询的所有表名, inputSchema{type: object, properties: {}, required: []}, ), Tool( namequery_oracle_table, description查询指定 Oracle 表的数据仅支持白名单中的表和列, inputSchema{ type: object, properties: { table_name: {type: string, description: 要查询的表名必须大写}, columns: { type: array, items: {type: string}, description: 要查询的列名列表为空则返回所有允许列, }, limit: {type: integer, description: 返回行数上限默认 10最大 100, default: 10}, filter_column: {type: string, description: 可选过滤列}, filter_value: {type: string, description: 可选过滤值}, }, required: [table_name], }, ), ] app.call_tool() async def call_tool(name: str, arguments: Dict[str, Any]) - List[TextContent]: try: if name list_allowed_tables: tables list(ALLOWED_TABLES.keys()) return [TextContent(typetext, textf允许查询的表: {, .join(tables)})] elif name query_oracle_table: table_name arguments.get(table_name, ).upper() requested_columns arguments.get(columns, []) limit min(arguments.get(limit, 10), 100) filter_col arguments.get(filter_column) filter_val arguments.get(filter_value) if table_name not in ALLOWED_TABLES: return [TextContent(typetext, textf错误: 表 {table_name} 不在白名单中)] allowed_cols ALLOWED_TABLES[table_name] if not requested_columns or * in requested_columns: select_cols allowed_cols else: invalid_cols [c for c in requested_columns if c.upper() not in allowed_cols] if invalid_cols: return [TextContent(typetext, textf错误: 列 {invalid_cols} 不允许查询)] select_cols [c.upper() for c in requested_columns] col_str , .join(select_cols) params {} base_sql fSELECT {col_str} FROM {table_name} if filter_col and filter_val is not None: if filter_col.upper() not in allowed_cols: return [TextContent(typetext, textf错误: 过滤列 {filter_col} 不允许)] base_sql f WHERE {filter_col.upper()} :bind_val params[:bind_val] filter_val sql fSELECT * FROM ({base_sql}) WHERE ROWNUM {limit} conn get_db_connection() cursor conn.cursor() try: cursor.execute(sql, params) columns [desc[0] for desc in cursor.description] rows cursor.fetchall() result_text f查询结果 {table_name}:\n result_text | | .join(columns) |\n result_text | |.join([--- for _ in columns]) |\n for row in rows: formatted [str(item) if item is not None else NULL for item in row] result_text | | .join(formatted) |\n return [TextContent(typetext, textresult_text)] finally: cursor.close() conn.close() else: return [TextContent(typetext, textf未知工具: {name})] except Exception as e: logger.error(f工具执行错误 {name}: {e}) return [TextContent(typetext, textf错误: {str(e)})] def main(): init_oracle_client() try: init_connection_pool() except Exception: logger.error(数据库连接池初始化失败MCP 服务不启动) return sse SseServerTransport(/messages/) async def handle_sse(request: Request): async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) async def handle_messages(request: Request): await sse.handle_post_message(request.scope, request.receive, request._send) return Response(status_code200) starlette_app Starlette( debugTrue, routes[ Route(/sse, endpointhandle_sse), Route(/messages/, endpointhandle_messages, methods[POST]), Route(/messages, endpointhandle_messages, methods[POST]), ], ) logger.info(启动 Oracle MCP Server端口 8080) uvicorn.run(starlette_app, host0.0.0.0, port8080) if __name__ __main__: main()环境变量文件/opt/mcp-oracle/.env这样写DB_HOST127.0.0.1 DB_PORT1521 DB_USERmcp_readonly DB_PASSWORD你的数据库密码 DB_SERVICEORCLPDB1 ORACLE_CLIENT_DIR/opt/oracle/instantclient_21_12 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的TaoToken密钥 TAOTOKEN_MODELclaude-sonnet-4-20250514Workbuddy 侧的 MCP 配置在设置里找到 MCP Servers粘贴下面这段 JSON。注意url里的 IP 换成你数据服务器的内网地址端口和代码里一致{ mcpServers: { Oracle-Connector: { transport: sse, url: http://10.0.1.50:8080/sse, disabled: false, headers: { Authorization: Bearer sk-你的TaoToken密钥 } } } }如果你用的是 Cline 或 Claude Code 这类支持 MCP 的客户端配置结构类似把transport和url对应填好即可。Codex 用户如果走auth.json把base_url指向https://taotoken.net/apiapi_key填同一个 Key。三件套Base URL Key Model ID在 MCP 服务端和客户端两侧保持一致后面排查会省很多事。4. 验证请求一次完整的连通性检查与成功结果配置写完了别急着在 Workbuddy 里问业务问题先做一次最小连通性验证。分三步MCP 服务端自检、SSE 端点探测、Workbuddy 工具调用。第一步在数据服务器上启动服务cd /opt/mcp-oracle set -a source .env set a python server.py看到启动 Oracle MCP Server端口 8080和Oracle 连接池初始化完成就说明服务端起来了。如果卡在连接池初始化检查数据库账号密码和 service_name。第二步在另一台能访问数据服务器的机器上探测 SSE 端点curl -N -H Accept: text/event-stream http://10.0.1.50:8080/sse正常会返回类似event: endpoint和data: /messages/?session_idxxx的内容连接保持不关闭。如果 curl 直接报Connection refused检查防火墙和 uvicorn 监听地址如果返回 404检查路由路径是不是/sse。第三步打开 Workbuddy在对话里明确指定使用 Oracle-Connector 这个 MCP。比如输入「用 Oracle-Connector 查询 SALES.ORDERS 表的前 5 行列出 ORDER_ID 和 AMOUNT」。Workbuddy 会先调用list_allowed_tables确认表在不在白名单再调用query_oracle_table带参数查询。成功的话你会看到类似下面的返回查询结果 SALES.ORDERS: | ORDER_ID | AMOUNT | |---|---| | 1001 | 299.00 | | 1002 | 1580.50 | | 1003 | 76.20 | | 1004 | 4300.00 | | 1005 | 215.75 |到这一步整条链路就通了Workbuddy 理解意图 → 通过 TaoToken 通道调用模型生成工具参数 → MCP 服务器执行受限 SQL → 返回 Markdown 表格 → 模型整理成自然语言回复。你可以再试一个带过滤条件的「查 SALES.CUSTOMERS 里 REGION 是 East 的客户只要 NAME 和 LEVEL 两列」验证白名单列过滤和 bind 参数是否生效。实测下来从启动服务到 Workbuddy 拿到第一份报表顺利的话十分钟内能跑通。真正花时间的是数据库账号权限配置和网络策略开通代码本身不复杂。5. 本篇常见错误排查401、local proxy failed 与 reading choices 报错部署过程中最容易撞上的几个报错我按出现频率排一下。401 UnauthorizedWorkbuddy 侧调用 MCP 时返回 401九成是Authorization头没带对。检查 JSON 配置里headers的 Key 是不是Authorization值是不是Bearer sk-xxxBearer 和 Key 之间有一个空格。另一个可能是 TaoToken 的 Key 过期或被删了去控制台重新生成一个换上。如果 MCP 服务端日志里看到模型调用 401检查.env里TAOTOKEN_API_KEY有没有被 shell 转义吃掉用echo $TAOTOKEN_API_KEY确认一下。local proxy failed这个报错通常出现在 Workbuddy 尝试连接 MCP 的 SSE 端点时。原因一般是网络不通——Workbuddy 所在机器访问不到数据服务器的 8080 端口。先在 Workbuddy 机器上telnet 10.0.1.50 8080测一下不通就找网络组开策略。还有一种情况是 MCP 服务端绑定了127.0.0.1而不是0.0.0.0外部访问不了检查uvicorn.run的 host 参数。reading choices 报错这个一般出现在模型返回结果解析阶段提示读取choices字段失败。根因通常是 TaoToken 通道返回的响应格式和客户端预期不一致。检查TAOTOKEN_BASE_URL是不是写成了https://taotoken.net/api/v1而客户端又自动拼了/v1导致路径变成/api/v1/v1/chat/completions。统一用https://taotoken.net/api让客户端自己拼版本路径。另外确认模型 ID 拼写正确不存在的模型 ID 有时会返回非标准错误体客户端解析时就报 reading choices。OAuth 相关报错如果你用的客户端走 OAuth 流程接 MCP报OAuth token invalid或redirect_uri mismatch先确认是不是把 API Key 认证和 OAuth 认证混用了。MCP 的 SSE 传输用 Bearer Token 就够了不需要走 OAuth。如果客户端强制 OAuth检查回调地址有没有在 TaoToken 控制台登记。数据库连接超时MCP 服务端日志报DPY-6005或连接池获取超时检查数据库监听是否正常、service_name是否写成了 SID。Oracle 的 service_name 和 SID 是两个东西用lsnrctl status确认一下。连接池max4在高并发下可能不够但内网小团队用足够了真不够再调。排查顺序建议先看 MCP 服务端日志再看 Workbuddy 侧报错最后用 curl 直接打 SSE 端点。日志里logger.error打出来的信息比客户端报错详细得多养成先看服务端日志的习惯。6. 把 MCP endpoint 统一到 TaoToken 后的日常维护通道统一之后日常维护就简单了。所有模型调用都经过 TaoToken你在控制台能看到每个 Key 的调用量和费用MCP 服务端和 Workbuddy 侧共用同一个 Key账单不会分散。要换模型改.env里的TAOTOKEN_MODEL重启服务就行Workbuddy 侧不用动。要加新的表白名单改ALLOWED_TABLES字典重启权限控制始终在 MCP 服务端这一层。API Key 的轮换也方便在控制台新建一个 Key把.env和 Workbuddy 配置里的旧 Key 替换掉重启 MCP 服务旧 Key 在控制台禁用。整个过程不影响数据库也不需要在多个工具之间同步修改。如果你后面要接更多数据源比如 MySQL 或 PostgreSQL可以再起一个 MCP 服务实例监听不同端口Workbuddy 侧加一个mcpServers条目就行。每个实例独立白名单、独立连接池互不干扰。模型通道还是同一个 TaoToken管理成本不随数据源数量线性增长。最后提醒一句MCP 服务端的白名单一定要认真配别图省事用SELECT *放开所有列。AI 工具再智能权限边界还是得靠代码守住。数据库账号用只读账号只授白名单表的 SELECT 权限双保险。