尧图网站建设 
1. 从 Function Calling 到 GraphRAG我踩过的 MCP 落地坑MCPModel Context Protocol是 Anthropic 开源的一套协议作用是把大语言模型直接接到数据源和外部工具上让模型能调用天气接口、数据库、知识库这些真实能力。它适合谁适合已经在用 LangChain、Qwen-Agent 做 Agent 开发但被 Function Calling 的每个工具都要手写 schema、每个客户端都要重新适配折磨过的同学。我这次把整条链路跑了一遍从最原始的 Function Calling 手写工具到用 MCP 统一封装成 Server再到接入 GraphRAG 做知识库问答最后用 TaoToken 统一管理 Key。整个过程踩了不少坑比如 MCP 的input_schema和 OpenAI 的parameters字段名不一致、SSE 模式下本机能访问但别的客户端连不上、GraphRAG 索引跑一半报 token 超限。这篇就把可复制的配置和验证动作完整记录下来你照着做基本能复现。先说清楚 MCP 和 Function Calling 的关系很多人一开始会混淆。Function Calling 是模型侧的能力模型训练时见过大量工具调用样本所以能识别外部工具并发起调用请求MCP 是工程侧的协议它把外部工具的运行脚本叫做 Server把接入这些工具的大模型运行环境叫做 Client。两者不冲突MCP Server 暴露的 Tools最终还是要转成 Function Calling 的格式喂给模型。所以你会看到我下面既有 MCP 的settings.json也有把input_schema转成parameters的转换函数。2. TaoToken 前置统一 Key 与接入地址在动手写代码前先把 Key 的事情理清楚。我这次所有模型调用都走 TaoToken 统一入口好处是不用在每个 Server、每个 Client 里散落不同的 base_url 和 api_key改一处就全生效。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要先去控制台创建一个 Key然后把它写进环境变量后面所有代码都从环境变量读不要硬编码。# .env 文件放在项目根目录 TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api/v1 CHAT_MODEL_NAME你的模型名这里有个细节OpenAI SDK 的base_url通常要带/v1后缀而 MCP 的settings.json里如果用的是openai_chat类型api_base也要带/v1。我一开始漏了/v1报了一堆 404排查了半小时才发现。Key 的创建入口在 https://taotoken.net/api-keys 接入文档在 https://taotoken.net/doc 遇到字段不确定的时候直接翻文档比猜快。注意不要把 Key 提交到 Git。.env一定要写进.gitignore团队协作时用.env.example占位。3. 可复制配置settings.json 与 config.toml 骨架MCP 在客户端里的标准接入方式就是写一个配置文件。不同客户端格式略有差异但核心字段一致command是要执行的程序args是参数env是环境变量。下面这个settings.json是我实测能跑通的骨架包含 filesystem、fetch、sqlite 三个常用 Server。{ mcpServers: { filesystem: { command: npx, args: [ -y, modelcontextprotocol/server-filesystem, /Users/yourname/projects/data ], env: {} }, fetch: { command: uvx, args: [mcp-server-fetch] }, sqlite: { command: uvx, args: [ mcp-server-sqlite, --db-path, ./test.db ] } } }command用npx表示这是个 Node 包用uvx表示这是个 Python 包。-y是 npx 的自动确认参数不加会卡在交互提示。args里第一个是包名后面是传给这个 Server 的参数比如 sqlite 的--db-path。如果你用的是支持 TOML 的客户端比如某些 CLI 工具等价配置长这样[mcp_servers.filesystem] command npx args [-y, modelcontextprotocol/server-filesystem, /Users/yourname/projects/data] [mcp_servers.fetch] command uvx args [mcp-server-fetch] [mcp_servers.sqlite] command uvx args [mcp-server-sqlite, --db-path, ./test.db]写完配置后客户端启动时会自动拉起这些子进程通过 stdio 建立 JSON-RPC 通信。你可以在客户端里看到已连接到服务器支持以下工具的日志说明配置生效了。接下来是 Python 侧的依赖安装。我用的清华源速度快很多uv pip install mcp langchain langchain-community langchain-openai chromadb httpx python-dotenv openai -i https://pypi.tuna.tsinghua.edu.cn/simple装完后mcp版本应该在 1.27 左右langchain1.2.xopenai2.x。版本差异会导致 API 变化比如mcp早期版本ClientSession的初始化方式和现在不同建议锁版本。4. 验证请求从 Function Calling 到 GraphRAG 检索4.1 先跑通最原始的 Function Calling在引入 MCP 之前先用最朴素的方式理解工具调用。下面这段代码定义了两个工具查天气和写文件然后让模型自己决定调哪个。from openai import OpenAI import json, os client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) def get_weather(loc): 查询即时天气loc 用英文城市名如 Beijing import requests url https://api.openweathermap.org/data/2.5/weather params {q: loc, appid: 你的key, units: metric, lang: zh_cn} return json.dumps(requests.get(url, paramsparams).json()) def write_file(content): with open(test.txt, w, encodingutf-8) as f: f.write(content) return 已成功写入本地文件。 available_functions {get_weather: get_weather, write_file: write_file} tools [ { type: function, function: { name: get_weather, description: 查询即时天气一次只能输入一个城市名, parameters: { type: object, properties: { loc: {type: string, description: 城市英文名如 Beijing} }, required: [loc] } } }, { type: function, function: { name: write_file, description: 将指定内容写入本地文件, parameters: { type: object, properties: { content: {type: string, description: 要写入的内容} }, required: [content] } } } ] def chat_base(messages): response client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages, toolstools ) if response.choices[0].finish_reason tool_calls: while True: messages.append(response.choices[0].message.model_dump()) for tc in response.choices[0].message.tool_calls: fn available_functions[tc.function.name] args json.loads(tc.function.arguments) result fn(**args) messages.append({ role: tool, content: result, tool_call_id: tc.id }) response client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages, toolstools ) if response.choices[0].finish_reason ! tool_calls: break return response messages [{role: user, content: 请问北京和上海今天天气如何并将这两个地点天气信息写入本地文档。}] resp chat_base(messages) print(resp.choices[0].message.content)跑通后你会看到模型先调get_weather两次再调write_file一次最后给出自然语言总结。这就是 Function Calling 的完整闭环用户提问 → 模型决定调工具 → 执行工具 → 结果喂回模型 → 模型继续思考 → 输出最终答案。4.2 用 MCP 把工具封装成 Server上面每个工具都要手写 schema工具一多就爆炸。MCP 的做法是把工具定义在 Server 里Client 通过协议自动发现。下面是一个天气 MCP Serverimport json, httpx, os from typing import Any from mcp.server.fastmcp import FastMCP mcp FastMCP(WeatherServer) OPENWEATHER_API_BASE https://api.openweathermap.org/data/2.5/weather API_KEY os.getenv(OPENWEATHER_API_KEY) async def fetch_weather(city: str) - dict[str, Any] | None: params {q: city, appid: API_KEY, units: metric, lang: zh_cn} async with httpx.AsyncClient() as client: try: resp await client.get(OPENWEATHER_API_BASE, paramsparams, timeout30.0) resp.raise_for_status() return resp.json() except Exception as e: return {error: str(e)} mcp.tool() async def query_weather(city: str) - str: 输入指定城市的英文名称返回今日天气查询结果。 data await fetch_weather(city) if error in data: return f查询失败: {data[error]} return f{data[name]} 温度 {data[main][temp]}°C湿度 {data[main][humidity]}% if __name__ __main__: mcp.run(transportstdio)Client 侧连接这个 Server并把 MCP 的input_schema转成 OpenAI 的parametersimport asyncio, json, os from contextlib import AsyncExitStack from openai import OpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class MCPClient: def __init__(self): self.exit_stack AsyncExitStack() self.client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) self.session None async def connect(self, script_path): params StdioServerParameters(commandpython, args[script_path], envNone) transport await self.exit_stack.enter_async_context(stdio_client(params)) self.stdio, self.write transport self.session await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() resp await self.session.list_tools() print(已连接工具列表:, [t.name for t in resp.tools]) async def process_query(self, query): messages [{role: user, content: query}] resp await self.session.list_tools() available_tools [{ type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema # 注意这里字段名转换 } } for t in resp.tools] response self.client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages, toolsavailable_tools ) content response.choices[0] if content.finish_reason tool_calls: tc content.message.tool_calls[0] args json.loads(tc.function.arguments) result await self.session.call_tool(tc.function.name, args) messages.append(content.message.model_dump()) messages.append({ role: tool, content: result.content[0].text, tool_call_id: tc.id }) response self.client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages ) return response.choices[0].message.content return content.message.content async def cleanup(self): await self.exit_stack.aclose() async def main(): client MCPClient() try: await client.connect(weather_server.py) answer await client.process_query(北京今天天气怎么样) print(answer) finally: await client.cleanup() if __name__ __main__: asyncio.run(main())关键点MCP 返回的工具 schema 字段叫inputSchema而 OpenAI 要的是parameters。我一开始直接传input_schema模型报缺少 parameters改成parameters就好了。这个转换在 Qwen-Agent 里是自动做的但手写 Client 时必须自己处理。4.3 接入 GraphRAG 做知识库检索GraphRAG 是微软开源的知识图谱增强检索方案它先把文档抽成实体和关系构建社区再基于社区报告做全局检索。安装uv pip install graphrag2.1.0 -i https://pypi.tuna.tsinghua.edu.cn/simple初始化项目并放入数据mkdir -p ./graphrag/input # 把你的 txt 文档放到 input 目录 graphrag init --root ./graphrag然后改settings.yaml重点是模型配置指向 TaoTokenmodels: default_chat_model: type: openai_chat api_base: https://taotoken.net/api/v1 auth_type: api_key api_key: ${TAOTOKEN_API_KEY} model: 你的模型名 encoding_model: cl100k_base model_supports_json: true concurrent_requests: 25 async_mode: threaded retry_strategy: native max_retries: -1 tokens_per_minute: 0 requests_per_minute: 0 default_embedding_model: type: openai_embedding api_base: https://taotoken.net/api/v1 auth_type: api_key api_key: ${TAOTOKEN_API_KEY} model: 你的embedding模型名 encoding_model: cl100k_base concurrent_requests: 25 async_mode: threaded retry_strategy: native max_retries: -1 tokens_per_minute: 0 requests_per_minute: 0 vector_store: default_vector_store: type: lancedb db_uri: output/lancedb container_name: default overwrite: True embed_text: model_id: default_embedding_model vector_store_id: default_vector_store input: type: file file_type: text base_dir: input chunks: size: 1200 overlap: 100 group_by_columns: [id] cache: type: file base_dir: cache reporting: type: file base_dir: logs output: type: file base_dir: output extract_graph: model_id: default_chat_model prompt: prompts/extract_graph.txt entity_types: [organization, person, geo, event] max_gleanings: 1 summarize_descriptions: model_id: default_chat_model prompt: prompts/summarize_descriptions.txt max_length: 500 community_reports: model_id: default_chat_model graph_prompt: prompts/community_report_graph.txt text_prompt: prompts/community_report_text.txt max_length: 2000 max_input_length: 8000 cluster_graph: max_cluster_size: 10 embed_graph: enabled: false umap: enabled: false snapshots: graphml: false embeddings: false local_search: chat_model_id: default_chat_model embedding_model_id: default_embedding_model prompt: prompts/local_search_system_prompt.txt global_search: chat_model_id: default_chat_model map_prompt: prompts/global_search_map_system_prompt.txt reduce_prompt: prompts/global_search_reduce_system_prompt.txt knowledge_prompt: prompts/global_search_knowledge_system_prompt.txt建索引graphrag index --root ./graphrag这一步会跑很久取决于文档量和模型速度。跑完后output/目录下会有entities.parquet、communities.parquet、community_reports.parquet三个文件。验证检索graphrag query --root ./graphrag --method global --query 对文档中的主要人物做一个详细介绍 graphrag query --root ./graphrag --method local --query 某个具体事件的细节global模式适合总结性问题local模式适合具体细节问题。我实测下来global 模式对整体脉络类问题效果好local 模式对某个人物做了什么这类问题更准。4.4 把 GraphRAG 封装成 MCP Server最后一步把 GraphRAG 检索能力也封装成 MCP Server这样任何支持 MCP 的客户端都能调用from pathlib import Path import pandas as pd import graphrag.api as api from graphrag.config.load_config import load_config from mcp.server.fastmcp import FastMCP mcp FastMCP(rag_ML) PROJECT_DIRECTORY /path/to/your/graphrag mcp.tool() async def rag_ML(query: str) - str: 用于查询文档中的人物或发生的事件 config load_config(Path(PROJECT_DIRECTORY)) entities pd.read_parquet(f{PROJECT_DIRECTORY}/output/entities.parquet) communities pd.read_parquet(f{PROJECT_DIRECTORY}/output/communities.parquet) community_reports pd.read_parquet(f{PROJECT_DIRECTORY}/output/community_reports.parquet) response, context await api.global_search( configconfig, entitiesentities, communitiescommunities, community_reportscommunity_reports, community_level2, dynamic_community_selectionFalse, response_typeMultiple Paragraphs, queryquery ) return response if __name__ __main__: mcp.run(transportstdio)Client 侧和 4.2 的代码完全一样只是把weather_server.py换成rag_server.py。这样你就有了一个能查天气、能查知识库的统一 Agent。5. 本篇常见错排查报错一input_schema字段不识别。这是最常见的。MCP 返回的工具定义里字段叫inputSchemaOpenAI 要的是parameters。手写 Client 时必须转换Qwen-Agent 会自动处理。转换逻辑就是取inputSchema的type、properties、required三个字段塞进parameters。报错二SSE 模式本机能访问其他客户端连不上。默认mcp.run(transportsse)只监听127.0.0.1外部访问不了。需要在启动时指定 host 和 port或者用反向代理转发。我试过直接改FastMCP的初始化参数但更稳的方式是在启动命令里加环境变量控制监听地址。报错三GraphRAG 索引跑到一半报 token 超限。把concurrent_requests从 25 降到 5chunk size从 1200 降到 800max_gleanings从 1 降到 0。这三个参数是 token 消耗大户降下来基本能跑完。报错四uvx找不到包。确认uv已安装且uvx在 PATH 里。如果用的是 conda 环境uvx可能不在当前环境需要pip install uv后重新激活。报错五MCP Server 启动后立刻退出。大概率是command写错了比如 Python 脚本用了node启动。检查command和args是否匹配文件类型。另外envNone表示继承当前进程环境变量如果 Server 需要特定 Key要么写进env要么确保当前 shell 已 export。报错六模型不调用工具直接回答。检查tools参数是否传了以及模型的finish_reason是不是tool_calls。有些模型对工具描述敏感description写得太模糊会导致模型忽略工具。把description写具体比如查询指定城市的实时天气输入英文城市名比查天气效果好很多。6. 语义一致 CTA整条链路跑下来最省事的做法是用 TaoToken 统一管理 Key避免在多个 Server 和 Client 里散落配置。如果你在接入过程中遇到 Key 相关的问题直接去 https://taotoken.net/api-keys 重新生成一个然后对照 https://taotoken.net/doc 检查字段格式。想先验证模型能不能正常对话可以用 https://taotoken.net/model-chat 快速测一下。如果是长期做编码和 Agent 开发建议了解一下 https://taotoken.net/coding-plan 比按量付费更划算。控制台在 https://taotoken.net/console 所有配置和用量都在那里看。 MCP 使用详细记录：从 Function Calling 到 GraphRAG 的配置与验证
1. 从 Function Calling 到 GraphRAG我踩过的 MCP 落地坑MCPModel Context Protocol是 Anthropic 开源的一套协议作用是把大语言模型直接接到数据源和外部工具上让模型能调用天气接口、数据库、知识库这些真实能力。它适合谁适合已经在用 LangChain、Qwen-Agent 做 Agent 开发但被 Function Calling 的每个工具都要手写 schema、每个客户端都要重新适配折磨过的同学。我这次把整条链路跑了一遍从最原始的 Function Calling 手写工具到用 MCP 统一封装成 Server再到接入 GraphRAG 做知识库问答最后用 TaoToken 统一管理 Key。整个过程踩了不少坑比如 MCP 的input_schema和 OpenAI 的parameters字段名不一致、SSE 模式下本机能访问但别的客户端连不上、GraphRAG 索引跑一半报 token 超限。这篇就把可复制的配置和验证动作完整记录下来你照着做基本能复现。先说清楚 MCP 和 Function Calling 的关系很多人一开始会混淆。Function Calling 是模型侧的能力模型训练时见过大量工具调用样本所以能识别外部工具并发起调用请求MCP 是工程侧的协议它把外部工具的运行脚本叫做 Server把接入这些工具的大模型运行环境叫做 Client。两者不冲突MCP Server 暴露的 Tools最终还是要转成 Function Calling 的格式喂给模型。所以你会看到我下面既有 MCP 的settings.json也有把input_schema转成parameters的转换函数。2. TaoToken 前置统一 Key 与接入地址在动手写代码前先把 Key 的事情理清楚。我这次所有模型调用都走 TaoToken 统一入口好处是不用在每个 Server、每个 Client 里散落不同的 base_url 和 api_key改一处就全生效。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要先去控制台创建一个 Key然后把它写进环境变量后面所有代码都从环境变量读不要硬编码。# .env 文件放在项目根目录 TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api/v1 CHAT_MODEL_NAME你的模型名这里有个细节OpenAI SDK 的base_url通常要带/v1后缀而 MCP 的settings.json里如果用的是openai_chat类型api_base也要带/v1。我一开始漏了/v1报了一堆 404排查了半小时才发现。Key 的创建入口在 https://taotoken.net/api-keys 接入文档在 https://taotoken.net/doc 遇到字段不确定的时候直接翻文档比猜快。注意不要把 Key 提交到 Git。.env一定要写进.gitignore团队协作时用.env.example占位。3. 可复制配置settings.json 与 config.toml 骨架MCP 在客户端里的标准接入方式就是写一个配置文件。不同客户端格式略有差异但核心字段一致command是要执行的程序args是参数env是环境变量。下面这个settings.json是我实测能跑通的骨架包含 filesystem、fetch、sqlite 三个常用 Server。{ mcpServers: { filesystem: { command: npx, args: [ -y, modelcontextprotocol/server-filesystem, /Users/yourname/projects/data ], env: {} }, fetch: { command: uvx, args: [mcp-server-fetch] }, sqlite: { command: uvx, args: [ mcp-server-sqlite, --db-path, ./test.db ] } } }command用npx表示这是个 Node 包用uvx表示这是个 Python 包。-y是 npx 的自动确认参数不加会卡在交互提示。args里第一个是包名后面是传给这个 Server 的参数比如 sqlite 的--db-path。如果你用的是支持 TOML 的客户端比如某些 CLI 工具等价配置长这样[mcp_servers.filesystem] command npx args [-y, modelcontextprotocol/server-filesystem, /Users/yourname/projects/data] [mcp_servers.fetch] command uvx args [mcp-server-fetch] [mcp_servers.sqlite] command uvx args [mcp-server-sqlite, --db-path, ./test.db]写完配置后客户端启动时会自动拉起这些子进程通过 stdio 建立 JSON-RPC 通信。你可以在客户端里看到已连接到服务器支持以下工具的日志说明配置生效了。接下来是 Python 侧的依赖安装。我用的清华源速度快很多uv pip install mcp langchain langchain-community langchain-openai chromadb httpx python-dotenv openai -i https://pypi.tuna.tsinghua.edu.cn/simple装完后mcp版本应该在 1.27 左右langchain1.2.xopenai2.x。版本差异会导致 API 变化比如mcp早期版本ClientSession的初始化方式和现在不同建议锁版本。4. 验证请求从 Function Calling 到 GraphRAG 检索4.1 先跑通最原始的 Function Calling在引入 MCP 之前先用最朴素的方式理解工具调用。下面这段代码定义了两个工具查天气和写文件然后让模型自己决定调哪个。from openai import OpenAI import json, os client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) def get_weather(loc): 查询即时天气loc 用英文城市名如 Beijing import requests url https://api.openweathermap.org/data/2.5/weather params {q: loc, appid: 你的key, units: metric, lang: zh_cn} return json.dumps(requests.get(url, paramsparams).json()) def write_file(content): with open(test.txt, w, encodingutf-8) as f: f.write(content) return 已成功写入本地文件。 available_functions {get_weather: get_weather, write_file: write_file} tools [ { type: function, function: { name: get_weather, description: 查询即时天气一次只能输入一个城市名, parameters: { type: object, properties: { loc: {type: string, description: 城市英文名如 Beijing} }, required: [loc] } } }, { type: function, function: { name: write_file, description: 将指定内容写入本地文件, parameters: { type: object, properties: { content: {type: string, description: 要写入的内容} }, required: [content] } } } ] def chat_base(messages): response client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages, toolstools ) if response.choices[0].finish_reason tool_calls: while True: messages.append(response.choices[0].message.model_dump()) for tc in response.choices[0].message.tool_calls: fn available_functions[tc.function.name] args json.loads(tc.function.arguments) result fn(**args) messages.append({ role: tool, content: result, tool_call_id: tc.id }) response client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages, toolstools ) if response.choices[0].finish_reason ! tool_calls: break return response messages [{role: user, content: 请问北京和上海今天天气如何并将这两个地点天气信息写入本地文档。}] resp chat_base(messages) print(resp.choices[0].message.content)跑通后你会看到模型先调get_weather两次再调write_file一次最后给出自然语言总结。这就是 Function Calling 的完整闭环用户提问 → 模型决定调工具 → 执行工具 → 结果喂回模型 → 模型继续思考 → 输出最终答案。4.2 用 MCP 把工具封装成 Server上面每个工具都要手写 schema工具一多就爆炸。MCP 的做法是把工具定义在 Server 里Client 通过协议自动发现。下面是一个天气 MCP Serverimport json, httpx, os from typing import Any from mcp.server.fastmcp import FastMCP mcp FastMCP(WeatherServer) OPENWEATHER_API_BASE https://api.openweathermap.org/data/2.5/weather API_KEY os.getenv(OPENWEATHER_API_KEY) async def fetch_weather(city: str) - dict[str, Any] | None: params {q: city, appid: API_KEY, units: metric, lang: zh_cn} async with httpx.AsyncClient() as client: try: resp await client.get(OPENWEATHER_API_BASE, paramsparams, timeout30.0) resp.raise_for_status() return resp.json() except Exception as e: return {error: str(e)} mcp.tool() async def query_weather(city: str) - str: 输入指定城市的英文名称返回今日天气查询结果。 data await fetch_weather(city) if error in data: return f查询失败: {data[error]} return f{data[name]} 温度 {data[main][temp]}°C湿度 {data[main][humidity]}% if __name__ __main__: mcp.run(transportstdio)Client 侧连接这个 Server并把 MCP 的input_schema转成 OpenAI 的parametersimport asyncio, json, os from contextlib import AsyncExitStack from openai import OpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class MCPClient: def __init__(self): self.exit_stack AsyncExitStack() self.client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) self.session None async def connect(self, script_path): params StdioServerParameters(commandpython, args[script_path], envNone) transport await self.exit_stack.enter_async_context(stdio_client(params)) self.stdio, self.write transport self.session await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() resp await self.session.list_tools() print(已连接工具列表:, [t.name for t in resp.tools]) async def process_query(self, query): messages [{role: user, content: query}] resp await self.session.list_tools() available_tools [{ type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema # 注意这里字段名转换 } } for t in resp.tools] response self.client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages, toolsavailable_tools ) content response.choices[0] if content.finish_reason tool_calls: tc content.message.tool_calls[0] args json.loads(tc.function.arguments) result await self.session.call_tool(tc.function.name, args) messages.append(content.message.model_dump()) messages.append({ role: tool, content: result.content[0].text, tool_call_id: tc.id }) response self.client.chat.completions.create( modelos.getenv(CHAT_MODEL_NAME), messagesmessages ) return response.choices[0].message.content return content.message.content async def cleanup(self): await self.exit_stack.aclose() async def main(): client MCPClient() try: await client.connect(weather_server.py) answer await client.process_query(北京今天天气怎么样) print(answer) finally: await client.cleanup() if __name__ __main__: asyncio.run(main())关键点MCP 返回的工具 schema 字段叫inputSchema而 OpenAI 要的是parameters。我一开始直接传input_schema模型报缺少 parameters改成parameters就好了。这个转换在 Qwen-Agent 里是自动做的但手写 Client 时必须自己处理。4.3 接入 GraphRAG 做知识库检索GraphRAG 是微软开源的知识图谱增强检索方案它先把文档抽成实体和关系构建社区再基于社区报告做全局检索。安装uv pip install graphrag2.1.0 -i https://pypi.tuna.tsinghua.edu.cn/simple初始化项目并放入数据mkdir -p ./graphrag/input # 把你的 txt 文档放到 input 目录 graphrag init --root ./graphrag然后改settings.yaml重点是模型配置指向 TaoTokenmodels: default_chat_model: type: openai_chat api_base: https://taotoken.net/api/v1 auth_type: api_key api_key: ${TAOTOKEN_API_KEY} model: 你的模型名 encoding_model: cl100k_base model_supports_json: true concurrent_requests: 25 async_mode: threaded retry_strategy: native max_retries: -1 tokens_per_minute: 0 requests_per_minute: 0 default_embedding_model: type: openai_embedding api_base: https://taotoken.net/api/v1 auth_type: api_key api_key: ${TAOTOKEN_API_KEY} model: 你的embedding模型名 encoding_model: cl100k_base concurrent_requests: 25 async_mode: threaded retry_strategy: native max_retries: -1 tokens_per_minute: 0 requests_per_minute: 0 vector_store: default_vector_store: type: lancedb db_uri: output/lancedb container_name: default overwrite: True embed_text: model_id: default_embedding_model vector_store_id: default_vector_store input: type: file file_type: text base_dir: input chunks: size: 1200 overlap: 100 group_by_columns: [id] cache: type: file base_dir: cache reporting: type: file base_dir: logs output: type: file base_dir: output extract_graph: model_id: default_chat_model prompt: prompts/extract_graph.txt entity_types: [organization, person, geo, event] max_gleanings: 1 summarize_descriptions: model_id: default_chat_model prompt: prompts/summarize_descriptions.txt max_length: 500 community_reports: model_id: default_chat_model graph_prompt: prompts/community_report_graph.txt text_prompt: prompts/community_report_text.txt max_length: 2000 max_input_length: 8000 cluster_graph: max_cluster_size: 10 embed_graph: enabled: false umap: enabled: false snapshots: graphml: false embeddings: false local_search: chat_model_id: default_chat_model embedding_model_id: default_embedding_model prompt: prompts/local_search_system_prompt.txt global_search: chat_model_id: default_chat_model map_prompt: prompts/global_search_map_system_prompt.txt reduce_prompt: prompts/global_search_reduce_system_prompt.txt knowledge_prompt: prompts/global_search_knowledge_system_prompt.txt建索引graphrag index --root ./graphrag这一步会跑很久取决于文档量和模型速度。跑完后output/目录下会有entities.parquet、communities.parquet、community_reports.parquet三个文件。验证检索graphrag query --root ./graphrag --method global --query 对文档中的主要人物做一个详细介绍 graphrag query --root ./graphrag --method local --query 某个具体事件的细节global模式适合总结性问题local模式适合具体细节问题。我实测下来global 模式对整体脉络类问题效果好local 模式对某个人物做了什么这类问题更准。4.4 把 GraphRAG 封装成 MCP Server最后一步把 GraphRAG 检索能力也封装成 MCP Server这样任何支持 MCP 的客户端都能调用from pathlib import Path import pandas as pd import graphrag.api as api from graphrag.config.load_config import load_config from mcp.server.fastmcp import FastMCP mcp FastMCP(rag_ML) PROJECT_DIRECTORY /path/to/your/graphrag mcp.tool() async def rag_ML(query: str) - str: 用于查询文档中的人物或发生的事件 config load_config(Path(PROJECT_DIRECTORY)) entities pd.read_parquet(f{PROJECT_DIRECTORY}/output/entities.parquet) communities pd.read_parquet(f{PROJECT_DIRECTORY}/output/communities.parquet) community_reports pd.read_parquet(f{PROJECT_DIRECTORY}/output/community_reports.parquet) response, context await api.global_search( configconfig, entitiesentities, communitiescommunities, community_reportscommunity_reports, community_level2, dynamic_community_selectionFalse, response_typeMultiple Paragraphs, queryquery ) return response if __name__ __main__: mcp.run(transportstdio)Client 侧和 4.2 的代码完全一样只是把weather_server.py换成rag_server.py。这样你就有了一个能查天气、能查知识库的统一 Agent。5. 本篇常见错排查报错一input_schema字段不识别。这是最常见的。MCP 返回的工具定义里字段叫inputSchemaOpenAI 要的是parameters。手写 Client 时必须转换Qwen-Agent 会自动处理。转换逻辑就是取inputSchema的type、properties、required三个字段塞进parameters。报错二SSE 模式本机能访问其他客户端连不上。默认mcp.run(transportsse)只监听127.0.0.1外部访问不了。需要在启动时指定 host 和 port或者用反向代理转发。我试过直接改FastMCP的初始化参数但更稳的方式是在启动命令里加环境变量控制监听地址。报错三GraphRAG 索引跑到一半报 token 超限。把concurrent_requests从 25 降到 5chunk size从 1200 降到 800max_gleanings从 1 降到 0。这三个参数是 token 消耗大户降下来基本能跑完。报错四uvx找不到包。确认uv已安装且uvx在 PATH 里。如果用的是 conda 环境uvx可能不在当前环境需要pip install uv后重新激活。报错五MCP Server 启动后立刻退出。大概率是command写错了比如 Python 脚本用了node启动。检查command和args是否匹配文件类型。另外envNone表示继承当前进程环境变量如果 Server 需要特定 Key要么写进env要么确保当前 shell 已 export。报错六模型不调用工具直接回答。检查tools参数是否传了以及模型的finish_reason是不是tool_calls。有些模型对工具描述敏感description写得太模糊会导致模型忽略工具。把description写具体比如查询指定城市的实时天气输入英文城市名比查天气效果好很多。6. 语义一致 CTA整条链路跑下来最省事的做法是用 TaoToken 统一管理 Key避免在多个 Server 和 Client 里散落配置。如果你在接入过程中遇到 Key 相关的问题直接去 https://taotoken.net/api-keys 重新生成一个然后对照 https://taotoken.net/doc 检查字段格式。想先验证模型能不能正常对话可以用 https://taotoken.net/model-chat 快速测一下。如果是长期做编码和 Agent 开发建议了解一下 https://taotoken.net/coding-plan 比按量付费更划算。控制台在 https://taotoken.net/console 所有配置和用量都在那里看。