1. 为什么要在本地把 Ollama 和 MCP 服务器接起来
如果你最近在折腾本地大模型，大概率会遇到一个尴尬：Ollama 跑起来了，聊天也正常，但模型只会“说”，不会“做”。你问它今天天气，它给你编一段；你让它查一下本地文件，它开始一本正经地胡说。这不是模型笨，而是它手里没有工具。
MCP（Model Context Protocol，模型上下文协议）解决的正是这件事。它把“工具”标准化成一种模型能理解的描述，模型输出结构化的调用意图，客户端负责真正执行。Ollama 负责推理，MCP 服务器负责提供工具，两者拼在一起，本地就有了一个能调用工具的 Agent 雏形。
这套组合适合谁？适合在个人开发机上想跑通工具调用链路的开发者。你不需要买 API 额度，不需要复杂集群，一台能跑 7B 左右模型的机器就够。我这次用的模型是gemma3:latest，工具是一个返回固定字符串的magicoutput，目的就是把链路跑通，而不是追求工具有多复杂。
链路本身分四段：MCP 服务器暴露工具、客户端拉取工具列表、把工具转成模型能懂的结构化 schema、模型返回调用意图后客户端真正执行。中间任何一段断了，你都会看到模型“假装调用”或者直接报错。
这里还有一个容易被忽略的点：本地 Ollama 默认只监听127.0.0.1:11434，MCP 服务器通过 stdio 启动，两者不在同一个进程里。所以你需要一个统一的入口来管理模型请求的 endpoint 和 Key，尤其是当你后面想换成远程模型或者多个模型混用时。TaoToken 在这里的角色就是统一 Key 和 endpoint，让你不用在每段代码里硬编码不同的地址。
我试过把 endpoint 写死在客户端里，换模型时改得满屏都是。后来统一走一个 base_url，代码干净很多。下面就从环境准备开始，一步步把这条链路搭起来。
2. TaoToken 前置准备：统一 Key 与 endpoint 设置
在写代码之前，先把“请求往哪发、用什么 Key”这件事定下来。本地 Ollama 的默认地址是http://localhost:11434，但如果你后面想接入更多模型，或者团队里多人共用一套配置，硬编码就不合适了。
TaoToken 提供统一的 API 入口，base_url 是https://taotoken.net/api。你需要在控制台创建一个 API Key，然后把它放进环境变量，而不是写进代码。这样做的好处是：换 Key 不用改代码，提交到 Git 也不会泄露。
先创建 Key。打开控制台页面，登录后进入 API Keys 管理，新建一个 Key，复制出来。地址是：
https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite拿到 Key 之后，在项目根目录建一个.env文件，写入两行：
TAOTOKEN_API_KEY=sk-你的实际key TAOTOKEN_BASE_URL=https://taotoken.net/api然后在 Python 里用os.environ读取。如果你用的是uv，它默认会加载.env，但为了保险，我在代码里显式读取一次。这里要注意，.env必须加进.gitignore，否则 Key 就跟着仓库走了。
接下来是模型 ID。Ollama 本地模型的 ID 就是你ollama list里看到的那个，比如gemma3:latest。如果你通过 TaoToken 走远程模型，模型 ID 用平台文档里列出的名称。两者在客户端里的写法不同：本地走ollama库的chat，远程走 OpenAI 兼容的 HTTP 请求。
为了统一，我在客户端里做了一个判断：如果TAOTOKEN_BASE_URL指向本地，就用ollama库；否则用requests发到 TaoToken 的 endpoint。这样同一份代码既能跑本地，也能切远程。
还有一个细节：MCP 服务器是通过 stdio 启动的子进程，它本身不关心模型 endpoint。模型 endpoint 只在客户端里用。所以你的配置只需要在客户端生效，服务器端保持干净。
配置检查清单：
| 配置项 | 本地 Ollama | TaoToken 远程 |
|---|---|---|
| base_url | http://localhost:11434 | https://taotoken.net/api |
| Key | 不需要 | 控制台创建 |
| 模型 ID | gemma3:latest | 平台文档名称 |
| 调用库 | ollama | requests / openai |
把这张表对照着填，后面排障时能省很多时间。配置好之后，就可以进入代码环节了。
3. 可复制配置：MCP 服务器与客户端完整代码
这一节是全文的核心，所有代码都可以直接复制运行。我按文件拆开讲，每个文件都给出完整内容，你只需要按顺序创建。
先建项目目录，然后初始化：
mkdir ollama-mcp-demo && cd ollama-mcp-demo uv init uv add fastmcp ollama mcp pydantic python-dotenvfastmcp是 MCP 服务器的实现库，ollama是本地模型客户端，mcp提供 stdio 连接能力，pydantic用来动态生成结构化 schema。装完之后目录结构应该是：
ollama-mcp-demo ├── server.py ├── client.py ├── .env └── pyproject.toml3.1 server.py：暴露一个可调用的工具
服务器端只做一件事：注册工具并启动。@mcp.tool()装饰器会把函数签名和 docstring 转成模型能读懂的 schema。
# server.py from fastmcp import FastMCP mcp = FastMCP("TestServer") @mcp.tool() def magicoutput(obj1: str, obj2: str) -> str: """使用这个函数来获取神奇的输出""" return f"WomboWombat:{obj1}:{obj2}" if __name__ == "__main__": mcp.run()注意返回值我改成了带参数的字符串，这样你能从最终结果里确认参数确实传进来了，而不是只看到一个固定值。docstring 很重要，模型就是靠它判断这个工具干什么用的。
3.2 client.py：拉取工具、转 schema、调用模型
客户端是重头戏。它要做四件事：启动 MCP 子进程、拉取工具列表、把工具转成 Pydantic 模型、把模型传给 Ollama 的format字段。
# client.py import asyncio import threading import queue import os from pathlib import Path from typing import Any, Optional, Union from dotenv import load_dotenv from pydantic import BaseModel, Field, create_model from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from ollama import chat load_dotenv() class OllamaMCP: def __init__(self, server_params: StdioServerParameters): self.server_params = server_params self.request_queue = queue.Queue() self.response_queue = queue.Queue() self.initialized = threading.Event() self.tools: list[Any] = [] self.response_model = None self.thread = threading.Thread(target=self._run_background, daemon=True) self.thread.start() def _run_background(self): asyncio.run(self._async_run()) async def _async_run(self): try: async with stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() self.session = session tools_result = await session.list_tools() self.tools = tools_result.tools self.initialized.set() while True: try: tool_name, arguments = self.request_queue.get(block=False) except queue.Empty: await asyncio.sleep(0.01) continue if tool_name is None: break try: result = await session.call_tool(tool_name, arguments) self.response_queue.put(result) except Exception as e: self.response_queue.put(f"错误: {str(e)}") except Exception as e: print("MCP 会话初始化错误:", str(e)) self.initialized.set() self.response_queue.put(f"MCP 初始化错误: {str(e)}") def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: if not self.initialized.wait(timeout=30): raise TimeoutError("MCP 会话初始化超时。") self.request_queue.put((tool_name, arguments)) return self.response_queue.get() def shutdown(self): self.request_queue.put((None, None)) self.thread.join() print("持久 MCP 会话已关闭。") @staticmethod def convert_json_type_to_python_type(json_type: str): if json_type == "integer": return (int, ...) if json_type == "number": return (float, ...) if json_type == "string": return (str, ...) if json_type == "boolean": return (bool, ...) return (str, ...) def create_response_model(self): dynamic_classes = {} for tool in self.tools: class_name = tool.name.capitalize() properties: dict[str, Any] = {} for prop_name, prop_info in tool.inputSchema.get("properties", {}).items(): json_type = prop_info.get("type", "string") properties[prop_name] = self.convert_json_type_to_python_type(json_type) model = create_model( class_name, __base__=BaseModel, __doc__=tool.description, **properties, ) dynamic_classes[class_name] = model if dynamic_classes: all_tools_type = Union[tuple(dynamic_classes.values())] Response = create_model( "Response", __base__=BaseModel, response=(str, Field(..., description="确认将调用该函数的消息")), tool=(all_tools_type, Field(..., description="用于运行并获取神奇输出的工具")), ) else: Response = create_model( "Response", __base__=BaseModel, response=(str, ...), tool=(Optional[Any], Field(None, description="如果不返回 None，则使用此工具")), ) self.response_model = Response async def ollama_chat(self, messages: list[dict[str, str]]) -> Any: conversation = [{ "role": "assistant", "content": f"你必须使用工具。你有以下函数可用。{[tool.name for tool in self.tools]}" }] conversation.extend(messages) if self.response_model is None: raise ValueError("响应模型尚未创建。") format_schema = self.response_model.model_json_schema() response = chat( model="gemma3:latest", messages=conversation, format=format_schema, ) print("Ollama 响应", response.message.content) response_obj = self.response_model.model_validate_json(response.message.content) maybe_tool = response_obj.tool if maybe_tool: function_name = maybe_tool.__class__.__name__.lower() func_args = maybe_tool.model_dump() output = await asyncio.to_thread(self.call_tool, function_name, func_args) return output return response_obj.response async def main(): server_parameters = StdioServerParameters( command="uv", args=["run", "python", "server.py"], cwd=str(Path.cwd()), ) persistent_session = OllamaMCP(server_parameters) if persistent_session.initialized.wait(timeout=30): print("已准备好调用工具。") else: print("错误：初始化超时。") persistent_session.create_response_model() messages = [ { "role": "system", "content": ( "你是一个听话的助手，上下文中有工具列表。" "你的任务是使用该函数来获取神奇的输出。" "不要自己生成神奇的输出。" "用简短的消息回复，提到将调用该函数，" "但不要提供函数输出本身。" "将该简短消息放在 'response' 属性中。" "同时在 'tool' 属性中填写正确的参数。" ), }, { "role": "user", "content": "使用该函数获取神奇的输出，参数为 (obj1 = Wombat 和 obj2 = Dog)", }, ] result = await persistent_session.ollama_chat(messages) print("最终结果:", result) persistent_session.shutdown() if __name__ == "__main__": asyncio.run(main())这段代码里有两个关键点。第一，format_schema直接来自 Pydantic 模型的model_json_schema()，Ollama 会用它约束输出结构。第二，工具名从类名反推，Magicoutput转成小写就是magicoutput，和服务器注册的名字对上。
如果你要换成 TaoToken 的远程模型，把chat(...)那段换成 HTTP 请求即可，base_url 用https://taotoken.net/api，Key 从环境变量读。模型 ID 换成平台文档里的名称。其余逻辑不变。
4. 验证请求：一次完整的工具调用与结果检查
代码写完之后，跑起来看结果。先确认 Ollama 服务在跑：
ollama serve另开一个终端，确认模型存在：
ollama list如果没看到gemma3:latest，先拉一个：
ollama pull gemma3:latest然后运行客户端：
uv run python client.py正常的话，你会看到类似下面的输出：
已准备好调用工具。 Ollama 响应 {"response": "好的，我将运行 magicoutput，参数为 obj1 = Wombat 和 obj2 = Dog。", "tool": {"obj1": "Wombat", "obj2": "Dog"}} 最终结果: meta=None content=[TextContent(type='text', text='WomboWombat:Wombat:Dog', annotations=None)] isError=False 持久 MCP 会话已关闭。检查点有三个。第一，Ollama 响应里必须是一个合法 JSON，包含response和tool两个字段。如果只有response没有tool，说明模型没按 schema 走，通常是系统提示不够明确，或者模型太小理解不了。第二，tool里的参数名必须和服务器函数签名一致，这里是obj1和obj2。第三，最终结果里的text应该包含你传进去的参数值，isError=False表示调用成功。
如果最终结果里出现isError=True，说明工具执行报错了，去看服务器端的异常信息。如果Ollama 响应直接是一段自然语言而不是 JSON，说明format字段没生效，检查format_schema是不是正确传进去了。
我实测下来，gemma3:latest对这种结构化输出支持不错，但如果你换成更小的模型，可能需要把系统提示写得更死板一些，比如明确说“只输出 JSON，不要输出其他内容”。
验证通过之后，你可以试着改一下用户消息里的参数，比如把Wombat换成Cat，再跑一次，确认参数确实透传到了服务器。这一步能帮你确认整条链路没有硬编码。
5. 本篇常见错误排查：401、local proxy failed、reading choices
链路跑通之前，大概率会踩几个坑。我把最常见的几个报错和对应解法列出来，你对照着看。
401 Unauthorized。这个通常出现在你切到 TaoToken 远程模型时。原因就两个：Key 没读到，或者 Key 无效。先检查.env里的TAOTOKEN_API_KEY是不是复制完整，有没有多余空格。然后在代码里打印一下os.environ.get("TAOTOKEN_API_KEY")的前几位，确认读到了。如果 Key 没问题，检查请求头是不是Authorization: Bearer sk-xxx，少个Bearer也会 401。
local proxy failed。这个报错一般出现在你配置了 HTTP 代理环境变量，但代理不可用的时候。本地 Ollama 请求走localhost，不应该经过代理。检查HTTP_PROXY和HTTPS_PROXY环境变量，如果设置了，在代码里对localhost做no_proxy排除。或者干脆在运行前unset HTTP_PROXY HTTPS_PROXY。
reading choices 报错。这个通常出现在你用 OpenAI 兼容接口调远程模型时，返回体里没有choices字段。原因可能是模型 ID 写错了，平台返回了一个错误对象而不是正常响应。先打印完整响应体，看error字段说了什么。如果是模型不存在，换成文档里列出的名称。
OAuth 相关报错。如果你用的是 Claude Code 或者某些需要 OAuth 的客户端，可能会看到 token 过期或者 scope 不足。这类问题不在本篇的 Ollama + MCP 链路里，但如果你后面要接 Claude Code，记得在配置里写全三件套：Base URL、API Key、Model ID。缺一个都会报 OAuth 或鉴权错误。
工具名对不上。表现是模型返回了tool字段，但客户端调用时报tool not found。原因是 Pydantic 类名转小写之后和服务器注册名不一致。比如服务器注册的是magic_output，类名会变成Magic_output，转小写是magic_output，能对上；但如果服务器用的是驼峰magicOutput，类名Magicoutput转小写就变成magicoutput，对不上。解决办法是统一用下划线命名。
初始化超时。initialized.wait(timeout=30)返回 False，说明 MCP 子进程没起来。检查server.py路径对不对，uv run python server.py能不能单独跑起来。如果单独跑报ModuleNotFoundError，说明依赖没装全，回到第 3 节重新uv add。
排障的时候，一个有用的技巧是在_async_run里把异常完整打印出来，不要只打印str(e)。很多错误信息在repr(e)里才完整。
6. 继续扩展：从单工具到 Agent 循环
链路跑通之后，你会发现这只是一个最小闭环。真正的 Agent 需要多轮循环：模型调用工具、拿到结果、再决定下一步。当前代码只做了一轮，模型返回工具意图后直接执行并结束。
要扩展成循环，思路是把ollama_chat里的结果再塞回对话历史，让模型基于工具返回值继续推理。具体做法是在call_tool拿到结果后，把结果作为tool角色的消息追加到messages，然后再次调用ollama_chat，直到模型不再返回tool字段为止。
另一个扩展方向是接真实的 MCP 服务器。现在你用的是自己写的magicoutput，但社区已经有很多现成的 MCP 服务器，比如文件系统、数据库查询、网页抓取。你只需要把server_parameters里的command和args换成对应服务器的启动命令，客户端逻辑完全不用改。这就是 MCP 标准化的价值。
如果你后面要接 Claude Code 或者 Cline 这类工具，配置里同样需要 Base URL、API Key、Model ID 三件套。Base URL 用https://taotoken.net/api，Key 从控制台拿，Model ID 按文档填。配置入口在：
https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite接入文档在这里，里面有不同客户端的配置示例：
https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite如果你想先验证模型对话是否正常，可以用模型对话页面快速测一下：
https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite长期跑编码类 Agent 的话，Coding Plan 会更合适，额度和模型选择都更偏向持续调用场景：
https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite最后说一个实用技巧：把server.py里的工具函数写成幂等的，也就是同样参数多次调用结果一致。Agent 循环里模型可能会重复调用同一个工具，幂等能避免副作用。另外，工具函数的 docstring 尽量写清楚参数含义和返回格式，模型理解得越准，调用成功率越高。
