从零开始构建 Ollama + MCP 服务器_ollama mcp

Model Context Protocol（模型上下文协议）在过去几个月里已经霸占了大家的视野，出现了许多酷炫的集成示例。我坚信它会成为一种标准，因为它正在定义工具与代理或软件与 AI 模型之间如何集成的新方式。

我决定尝试将 Ollama 中的一个小型 LLM 连接到一个 MCP 服务器上，感受一下这个新标准的魅力。今天，我想向大家展示一种将 Ollama 与 MCP 服务器集成的可能实现方式。

魔法配方

集成的主要步骤如下：

1. 创建一个测试用的 MCP 服务器。
2. 创建一个客户端文件，用于发送请求并启动服务器。
3. 将服务器的工具获取到客户端。
4. 将工具转换为 Pydantic 模型。
5. 通过响应的 format 字段将工具（作为 Pydantic 模型）传递给 Ollama。
6. 通过 Ollama 发送对话并接收结构化输出。
7. 如果响应中包含工具，则向服务器发起请求。

安装依赖

要运行这个项目，需要安装所需的包。fastmcp 库在使用 uv 运行代码时表现最佳。它像 Poetry 和 pip 一样，下载方便，使用简单。

使用以下命令将所需的库添加到项目中：

uv add fastmcp ollama

这将安装 MCP 服务器和 Ollama 聊天库，你可以在此基础上构建客户端和服务器逻辑。

文件结构

设置完成后，你的文件夹结构应该如下所示：

your folder├── server.py└── client.py

server.py 文件包含 MCP 服务器和你想要暴露的工具。client.py 文件在后台进程启动服务器，获取可用工具，并与 Ollama 连接。

示例 MCP 服务器

让我们从创建一个简单的 MCP 服务器开始，使用 fastmcp 库。服务器暴露了一个名为 magicoutput 的工具。该函数接受两个字符串输入，并返回一个固定的字符串作为输出。

使用 @mcp.tool() 装饰器将函数注册为 MCP 服务器中的可用工具。服务器启动后，任何客户端都可以获取并调用这个工具。

通过在主块中调用 mcp.run() 启动服务器。

# server.pyfrom fastmcp import FastMCP# 创建一个 MCP 服务器mcp = FastMCP(\"TestServer\")# 我的工具：@mcp.tool()def magicoutput(obj1: str, obj2: str) -> int: \"\"\"使用这个函数来获取神奇的输出\"\"\" return \"WomboWombat\"if __name__ == \"__main__\": mcp.run()

获取服务器工具

为了连接到 MCP 服务器并列出可用工具，我们使用 ClientSession、StdioServerParameters 和 stdio_client，它们都来自 mcp 库。

我们定义了一个名为 OllamaMCP 的类，用于处理服务器连接和工具获取。在类中，_async_run 方法启动一个异步会话，初始化它，并从服务器获取工具列表。

我们使用 threading.Event() 来跟踪会话何时准备就绪，并将工具列表存储在 self.tools 中。

在脚本的末尾，我们定义了服务器参数，并在后台线程中运行客户端。这将启动连接并打印服务器返回的工具元数据。

# client.pyimport asyncioimport threadingfrom pathlib import Pathfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom typing import Anyclass OllamaMCP: \"\"\" Ollama 和 FastMCP 之间的简单集成 \"\"\" def __init__(self, server_params: StdioServerParameters): self.server_params = server_params self.initialized = threading.Event() self.tools: list[Any] = [] 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  print(tools_result) except Exception as e: print(f\"初始化 MCP 服务器时出错 {str(e)}\")if __name__ == \"__main__\": server_parameters = StdioServerParameters( command=\"uv\", args=[\"run\", \"python\", \"server.py\"], cwd=str(Path.cwd()) ) ollamamcp = OllamaMCP(server_params=server_parameters) ollamamcp._run_background()

运行上述代码后，你会从服务器收到以下响应，其中可以看到服务器上可用的工具列表。

[04/14/25 22:29:08] INFO 正在启动服务器 \"TestServer\"... server.py:171  INFO 正在处理请求类型 server.py:534 ListToolsRequestmeta=None nextCursor=None tools=[Tool(name=\'magicoutput\', description=\'使用这个函数来获取神奇的输出\', inputSchema={\'properties\': {\'obj1\': {\'title\': \'Obj1\', \'type\': \'string\'}, \'obj2\': {\'title\': \'Obj2\', \'type\': \'string\'}}, \'required\': [\'obj1\', \'obj2\'], \'title\': \'magicoutputArguments\', \'type\': \'object\'})]

将工具转换为 Pydantic 模型

现在我们已经从服务器获取了工具列表，下一步是将它们转换为 Pydantic 模型。我们使用 Pydantic 的 create_model 动态定义一个新的响应模式，基于服务器的工具定义。还有一个辅助函数，用于将 JSON 类型映射为有效的 Python 类型。

这可以帮助我们动态定义模型，以便 LLM 精确地知道在返回工具参数时应使用什么结构。

# client.pyimport asyncioimport threadingfrom pathlib import Pathfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom typing import Any, Union, Optionalfrom pydantic import BaseModel, create_model, Fieldclass OllamaMCP: \"\"\"Ollama 和 FastMCP 之间的简单集成\"\"\" def __init__(self, server_params: StdioServerParameters): self.server_params = server_params self.initialized = threading.Event() self.tools: list[Any] = [] 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 except Exception as e: print(f\"初始化 MCP 服务器时出错 {str(e)}\") def create_response_model(self): dynamic_classes = {} for tool in self.tools: class_name = tool.name.capitalize() properties = {} 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, __doc__=\"LLm 响应类\", response=(str, Field(..., description= \"确认将调用该函数的消息\")), tool=(all_tools_type, Field(  ...,  description=\"用于运行并获取神奇输出的工具\" )), ) else: Response = create_model( \"Response\", __base__=BaseModel, __doc__=\"LLm 响应类\", response=(str, ...), tool=(Optional[Any], Field(None, description=\"如果不返回 None，则使用此工具\")), ) self.response_model = Response print(Response.model_fields) @staticmethod def convert_json_type_to_python_type(json_type: str): \"\"\"将 JSON 类型简单映射为 Python（Pydantic）类型\"\"\" 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, ...)# 从零开始构建 Ollama + MCP 服务器Model Context Protocol（模型上下文协议）在过去几个月里已经霸占了大家的视野，出现了许多酷炫的集成示例。我坚信它会成为一种标准，因为它正在定义工具与代理或软件与 AI 模型之间如何集成的新方式。既然我最喜欢写的是小型语言模型（LLMs），我决定尝试将 Ollama 中的一个小型 LLM 连接到一个 MCP 服务器上，感受一下这个新标准的魅力。今天，我想向大家展示一种将 Ollama 与 MCP 服务器集成的可能实现方式。## 魔法配方集成的主要步骤如下：1. 创建一个测试用的 MCP 服务器。2. 创建一个客户端文件，用于发送请求并启动服务器。3. 将服务器的工具获取到客户端。4. 将工具转换为 Pydantic 模型。5. 通过响应的 `format` 字段将工具（作为 Pydantic 模型）传递给 Ollama。6. 通过 Ollama 发送对话并接收结构化输出。7. 如果响应中包含工具，则向服务器发起请求。## 安装依赖要运行这个项目，需要安装所需的包。`fastmcp` 库在使用 `uv` 运行代码时表现最佳。它像 Poetry 和 pip 一样，下载方便，使用简单。使用以下命令将所需的库添加到项目中：```csharpuv add fastmcp ollama

这将安装 MCP 服务器和 Ollama 聊天库，你可以在此基础上构建客户端和服务器逻辑。

文件结构

设置完成后，你的文件夹结构应该如下所示：

your folder├── server.py└── client.py

server.py 文件包含 MCP 服务器和你想要暴露的工具。client.py 文件在后台进程启动服务器，获取可用工具，并与 Ollama 连接。

示例 MCP 服务器

让我们从创建一个简单的 MCP 服务器开始，使用 fastmcp 库。服务器暴露了一个名为 magicoutput 的工具。该函数接受两个字符串输入，并返回一个固定的字符串作为输出。

使用 @mcp.tool() 装饰器将函数注册为 MCP 服务器中的可用工具。服务器启动后，任何客户端都可以获取并调用这个工具。

通过在主块中调用 mcp.run() 启动服务器。

# server.pyfrom fastmcp import FastMCP# 创建一个 MCP 服务器mcp = FastMCP(\"TestServer\")# 我的工具：@mcp.tool()def magicoutput(obj1: str, obj2: str) -> int: \"\"\"使用这个函数来获取神奇的输出\"\"\" return \"WomboWombat\"if __name__ == \"__main__\": mcp.run()

获取服务器工具

为了连接到 MCP 服务器并列出可用工具，我们使用 ClientSession、StdioServerParameters 和 stdio_client，它们都来自 mcp 库。

我们定义了一个名为 OllamaMCP 的类，用于处理服务器连接和工具获取。在类中，_async_run 方法启动一个异步会话，初始化它，并从服务器获取工具列表。

我们使用 threading.Event() 来跟踪会话何时准备就绪，并将工具列表存储在 self.tools 中。

在脚本的末尾，我们定义了服务器参数，并在后台线程中运行客户端。这将启动连接并打印服务器返回的工具元数据。

# client.pyimport asyncioimport threadingfrom pathlib import Pathfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom typing import Anyclass OllamaMCP: \"\"\" Ollama 和 FastMCP 之间的简单集成 \"\"\" def __init__(self, server_params: StdioServerParameters): self.server_params = server_params self.initialized = threading.Event() self.tools: list[Any] = [] 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  print(tools_result) except Exception as e: print(f\"初始化 MCP 服务器时出错 {str(e)}\")if __name__ == \"__main__\": server_parameters = StdioServerParameters( command=\"uv\", args=[\"run\", \"python\", \"server.py\"], cwd=str(Path.cwd()) ) ollamamcp = OllamaMCP(server_params=server_parameters) ollamamcp._run_background()

运行上述代码后，你会从服务器收到以下响应，其中可以看到服务器上可用的工具列表。

[04/14/25 22:29:08] INFO 正在启动服务器 \"TestServer\"... server.py:171  INFO 正在处理请求类型 server.py:534 ListToolsRequestmeta=None nextCursor=None tools=[Tool(name=\'magicoutput\', description=\'使用这个函数来获取神奇的输出\', inputSchema={\'properties\': {\'obj1\': {\'title\': \'Obj1\', \'type\': \'string\'}, \'obj2\': {\'title\': \'Obj2\', \'type\': \'string\'}}, \'required\': [\'obj1\', \'obj2\'], \'title\': \'magicoutputArguments\', \'type\': \'object\'})]

将工具转换为 Pydantic 模型

现在我们已经从服务器获取了工具列表，下一步是将它们转换为 Pydantic 模型。我们使用 Pydantic 的 create_model 动态定义一个新的响应模式，基于服务器的工具定义。还有一个辅助函数，用于将 JSON 类型映射为有效的 Python 类型。

这可以帮助我们动态定义模型，以便 LLM 精确地知道在返回工具参数时应使用什么结构。

# client.pyimport asyncioimport threadingfrom pathlib import Pathfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom typing import Any, Union, Optionalfrom pydantic import BaseModel, create_model, Fieldclass OllamaMCP: \"\"\"Ollama 和 FastMCP 之间的简单集成\"\"\" def __init__(self, server_params: StdioServerParameters): self.server_params = server_params self.initialized = threading.Event() self.tools: list[Any] = [] 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 except Exception as e: print(f\"初始化 MCP 服务器时出错 {str(e)}\") def create_response_model(self): dynamic_classes = {} for tool in self.tools: class_name = tool.name.capitalize() properties = {} 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, __doc__=\"LLm 响应类\", response=(str, Field(..., description= \"确认将调用该函数的消息\")), tool=(all_tools_type, Field(  ...,  description=\"用于运行并获取神奇输出的工具\" )), ) else: Response = create_model( \"Response\", __base__=BaseModel, __doc__=\"LLm 响应类\", response=(str, ...), tool=(Optional[Any], Field(None, description=\"如果不返回 None，则使用此工具\")), ) self.response_model = Response print(Response.model_fields) @staticmethod def convert_json_type_to_python_type(json_type: str): \"\"\"将 JSON 类型简单映射为 Python（Pydantic）类型\"\"\" 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, ...)

运行代码后，print(Response.model_fields) 的输出将显示我们刚刚构建的响应模型的完整结构。该模型包括两部分：一部分是助手发送回用户的消息，另一部分是可选字段，用于保存工具参数。

如果模型填写了 tool 字段，我们将使用它来调用服务器。否则，我们只使用纯响应字符串。

uv run python -m convert_tools[04/15/25 10:15:32] INFO 正在启动服务器 \"TestServer\"... server.py:171  INFO 正在处理请求类型 server.py:534 ListToolsRequest{\'response\': FieldInfo(annotation=str, required=True, description=\'确认将调用该函数的消息\'), \'tool\': FieldInfo(annotation=Magicoutput, required=True, description=\'用于运行并获取神奇输出的工具\')}

使用后台线程和队列调用工具

现在工具已经作为 Pydantic 模型可用，我们可以继续启用工具调用。为此，我们使用一个后台线程并设置两个队列。一个用于向服务器发送请求，另一个用于接收响应。

call_tool 方法将请求放入队列中，后台线程监听该请求。一旦使用 MCP 会话调用工具，结果将放入响应队列中。

import asyncioimport threadingimport queuefrom pathlib import Pathfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom typing import Any, Union, Optionalfrom pydantic import BaseModel, create_model, Fieldclass OllamaMCP: \"\"\"Ollama 和 FastMCP 之间的简单集成\"\"\" def __init__(self, server_params: StdioServerParameters): self.server_params = server_params self.initialized = threading.Event() self.tools: list[Any] = [] self.request_queue = queue.Queue() self.response_queue = queue.Queue() # 启动后台线程以异步处理请求。 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: print(\"收到关闭信号。\") 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)) result = self.response_queue.get() return result def shutdown(self): \"\"\" 清理并关闭持久会话。 \"\"\" self.request_queue.put((None, None)) self.thread.join() print(\"持久 MCP 会话已关闭。\") def create_response_model(self): dynamic_classes = {} for tool in self.tools: class_name = tool.name.capitalize() properties = {} 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, ...), tool=(Optional[all_tools_type], Field(None, description=\"如果不返回 None，则使用此工具\")), ) else: Response = create_model( \"Response\", __base__=BaseModel, response=(str, ...), tool=(Optional[Any], Field(None, description=\"如果不返回 None，则使用此工具\")), ) self.response_model = Response @staticmethod def convert_json_type_to_python_type(json_type: str): \"\"\"将 JSON 类型简单映射为 Python（Pydantic）类型\"\"\" 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, ...)if __name__ == \"__main__\": server_parameters = StdioServerParameters( command=\"uv\", args=[\"run\", \"python\", \"server.py\"], cwd=str(Path.cwd()) ) ollamamcp = OllamaMCP(server_params=server_parameters) if ollamamcp.initialized.wait(timeout=30): print(\"已准备好调用工具。\") result = ollamamcp.call_tool( tool_name=\"magicoutput\", arguments={\"obj1\": \"dog\", \"obj2\": \"cat\"} ) print(result) else: print(\"错误：初始化超时。\")

请注意，此时我们是手动传递函数名和参数，使用 call_tool 方法。在下一节中，我们将根据 Ollama 返回的结构化输出触发此调用。

运行此代码后，我们可以确认一切按预期工作。工具被服务器正确识别、执行，并返回结果。

[04/15/25 11:37:47] INFO 正在启动服务器 \"TestServer\"... server.py:171  INFO 正在处理请求类型 server.py:534 ListToolsRequest已准备好调用工具。  INFO 正在处理请求类型 server.py:534 CallToolRequestmeta=None content=[TextContent(type=\'text\', text=\'WomboWombat\', annotations=None)] isError=False

Ollama + MCP

随着队列和 call_tool 函数准备就绪，现在是时候集成 Ollama 了。我们将响应类传递给 Ollama 的 format 字段，告诉我们的 LLM（这里使用 Gemma）在生成输出时遵循该模式。

我们还定义了一个 ollama_chat 方法，用于发送对话，验证模型的响应是否符合模式，并检查是否包含工具。如果是，则提取函数名和参数，然后使用持久的 MCP 会话在后台线程中调用它。

在 main 函数中，我们设置服务器连接，启动后台循环，并等待一切准备就绪。然后我们准备一个系统提示和用户消息，将它们发送到 Ollama，并等待结构化输出。

最后，我们打印服务器的结果并关闭会话。

import asyncioimport threadingimport queuefrom pathlib import Pathfrom typing import Any, Optional, Unionfrom pydantic import BaseModel, Field, create_modelfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom ollama import chatclass 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.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)) result = self.response_queue.get() return result 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): \"\"\"将 JSON 类型简单映射为 Python（Pydantic）类型\"\"\" 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): \"\"\" 根据获取的工具动态创建 Pydantic 响应模型。 \"\"\" 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, __doc__=\"LLm 响应类\", response=(str, Field(..., description= \"确认将调用该函数的消息\")), tool=(all_tools_type, Field(  ...,  description=\"用于运行并获取神奇输出的工具\" )), ) else: Response = create_model( \"Response\", __base__=BaseModel, __doc__=\"LLm 响应类\", response=(str, ...), tool=(Optional[Any], Field(None, description=\"如果不返回 None，则使用此工具\")), ) self.response_model = Response async def ollama_chat(self, messages: list[dict[str, str]]) -> Any: \"\"\" 使用动态响应模型向 Ollama 发送消息。 如果响应中包含工具，则使用持久会话调用它。 \"\"\" conversation = [{\"role\":\"assistant\", \"content\": f\"你必须使用工具。你有以下函数可用。{[ tool.name for tool in self.tools ]}\"}] conversation.extend(messages) if self.response_model is None: raise ValueError(\"响应模型尚未创建。请先调用 create_response_model()。\") # 获取聊天消息格式的 JSON 模式。 format_schema = self.response_model.model_json_schema() # 调用 Ollama（假设是同步的）并解析响应。 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() # 使用 asyncio.to_thread 在线程中调用同步的 call_tool 方法。 output = await asyncio.to_thread(self.call_tool, function_name, func_args) return output else: print(\"响应中未检测到工具。返回纯文本响应。\") return response_obj.responseasync 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() # 为 Ollama 准备消息。 messages = [ { \"role\": \"system\", \"content\": ( \"你是一个听话的助手，上下文中有工具列表。 \" \"你的任务是使用该函数来获取神奇的输出。 \" \"不要自己生成神奇的输出。 \" \"用简短的消息回复，提到将调用该函数， \" \"但不要提供函数输出本身。 \" \"将该简短消息放在 \'response\' 属性中。 \" \"例如：\'好的，我将运行 magicoutput 函数并返回输出。\' \" \"同时在 \'tool\' 属性中填写正确的参数。 \" ) }, { \"role\": \"user\", \"content\": \"使用该函数获取神奇的输出，参数为 (obj1 = Wombat 和 obj2 = Dog)\" } ] # 调用 Ollama 并处理响应。 result = await persistent_session.ollama_chat(messages) print(\"最终结果:\", result) # 关闭持久会话。 persistent_session.shutdown()if __name__ == \"__main__\": asyncio.run(main())

你可以看到输出完美无缺。我们收到了一个包含简短消息的 response，以及将发送到 MCP 服务器的 tool 和参数。最后，我们得到了服务器的输出，如下所示：

[04/15/25 09:52:49] INFO 正在启动服务器 \"TestServer\"... server.py:171  INFO 正在处理请求类型 server.py:534 ListToolsRequest已准备好调用工具。Ollama 响应 {\"response\": \"好的，我将运行 magicoutput，参数为 obj1 = Wombat 和 obj2 = Dog。\",\"tool\": {\"obj1\": \"Wombat\", \"obj2\": \"Dog\"}}[04/15/25 09:52:52] INFO 正在处理请求类型 server.py:534 CallToolRequest最终结果: meta=None content=[TextContent(type=\'text\', text=\'WomboWombat\', annotations=None)] isError=False持久 MCP 会话已关闭。

更多精彩内容

如果你真的喜欢这些博客，我敢肯定你也会喜欢接下来的内容。一定要为它们点赞，支持每篇博客所需的时间和努力！

总结

就这样，我们刚刚完成了一种将 Ollama 中的小型模型与本地 MCP 服务器连接的方法。

我们从创建一个简单的 MCP 服务器开始，它包含一个工具。然后我们构建了一个客户端，用于连接到服务器，获取工具定义，并将它们转换为 Pydantic 模型。在此基础上，我们通过 format 字段将响应模型传递给 Ollama。模型返回了一个结构化响应，我们使用后台线程和队列在客户端处理工具调用。

点赞并关注我，获取更多类似内容。后期将扩展这个功能，构建一个完整的代理循环，并开始使用真实的 MCP 服务器创建有用的工作流。