概述

在现代应用开发中，将大语言模型（LLM）与专用工具服务相结合，可以构建出既能理解自然语言，又能准确执行专业任务的智能代理。本文介绍一个基于 MCP（Model Context Protocol）协议和 Ollama 本地大模型的时间查询系统，它能够智能识别用户查询意图，并动态调用时间服务工具来提供准确的时区时间信息。

系统架构

该系统由两个核心组件构成：

时间服务器 (time_server.py)：基于 MCP 协议实现的专用时间服务，提供获取当前时间和列出常见时区的工具函数
客户端测试程序 (time_client_test3.py)：使用 Ollama 本地大模型分析用户查询，智能决定是否需要调用时间服务工具

核心代码解析

时间服务器实现

from mcp.server.fastmcp import FastMCP
from datetime import datetime
import pytz  # 用于处理时区，如果需要的话可以先安装：uv add pytz / pip install pytz# 创建 MCP 服务器实例，命名为 "TimeServer"
mcp = FastMCP("TimeServer")@mcp.tool()
def get_current_time(timezone: str = "UTC") -> str:"""获取指定时区的当前时间。参数:timezone (str): 时区名称，例如 'Asia/Shanghai', 'UTC', 'America/New_York'。默认为 'UTC'。返回:str: 格式化后的当前时间字符串，包含时区信息。"""try:# 获取指定时区tz = pytz.timezone(timezone)# 获取该时区的当前时间now = datetime.now(tz)# 格式化时间字符串formatted_time = now.strftime("%Y-%m-%d %H:%M:%S %Z")return f"当前时间 ({timezone}) 是: {formatted_time}"except pytz.UnknownTimeZoneError:return f"错误：未知的时区 '{timezone}'。请提供有效的时区名称，例如 'Asia/Shanghai', 'UTC'。"# 可选：再添加一个工具，获取所有支持的时区列表（例如列出一些常见的）
@mcp.tool()
def list_common_timezones() -> list:"""获取常见的时区列表。返回:list: 包含常见时区名称的列表。"""common_timezones = ['UTC', 'Asia/Shanghai', 'Asia/Tokyo', 'America/New_York', 'Europe/London', 'Australia/Sydney']return common_timezonesif __name__ == "__main__":# 运行 MCP 服务器mcp.run(transport='stdio')

关键特性：
● 使用 @mcp.tool() 装饰器将函数暴露为 MCP 工具
● 支持动态时区处理，使用 pytz 库处理全球时区
● 包含完整的错误处理机制，对未知时区提供友好提示

客户端智能代理

import asyncio
import json
import re
from typing import Dict, Any, Optional, Tuple
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import ollamaclass OllamaMCPClient:def __init__(self, server_script_path: str):self.server_params = StdioServerParameters(command="python",args=[server_script_path],)self.available_tools = []self.tool_info = {}self._session = Noneself._stdio_ctx = Noneasync def __aenter__(self):"""异步上下文管理器入口 - 初始化连接"""try:# 创建 stdio 客户端上下文self._stdio_ctx = stdio_client(self.server_params)read_stream, write_stream = await self._stdio_ctx.__aenter__()# 创建会话self._session = ClientSession(read_stream, write_stream)await self._session.initialize()# 获取可用工具列表tools_response = await self._session.list_tools()self.available_tools = [tool.name for tool in tools_response.tools]print(f"可用工具: {self.available_tools}")# 获取每个工具的详细信息for tool in tools_response.tools:self.tool_info[tool.name] = {"description": tool.description,"inputSchema": tool.inputSchema}return selfexcept Exception as e:await self.__aexit__(None, None, None)raise Exception(f"初始化 MCP 服务器时出错: {e}")async def __aexit__(self, exc_type, exc_val, exc_tb):"""异步上下文管理器出口 - 清理资源"""try:if self._session:await self._session.close()if self._stdio_ctx:await self._stdio_ctx.__aexit__(exc_type, exc_val, exc_tb)except Exception as e:print(f"关闭连接时出错: {e}")def extract_tool_call(self, model_response: str) -> Tuple[Optional[str], Optional[Dict[str, Any]]]:"""从模型响应中提取工具调用信息。使用多种策略处理不稳定的参数格式。"""# 策略1: 尝试解析整个响应为JSONtry:data = json.loads(model_response.strip())if isinstance(data, dict) and "name" in data:tool_name = data["name"]arguments = data.get("arguments", {})if tool_name in self.available_tools:validated_args = self.validate_arguments(tool_name, arguments)return tool_name, validated_argsexcept json.JSONDecodeError:pass# 策略2: 尝试查找JSON片段json_patterns = [r'\{[^{}]*"name"[^{}]*:[^{}]*"[^"]*"[^{}]*\}',r'\{.*"name":\s*"([^"]+)".*"arguments":\s*(\{.*?\}).*\}',]for pattern in json_patterns:matches = re.finditer(pattern, model_response, re.DOTALL)for match in matches:try:if match.lastindex == 2:tool_name = match.group(1)arguments = json.loads(match.group(2))else:data = json.loads(match.group(0))tool_name = data.get("name")arguments = data.get("arguments", {})if tool_name and tool_name in self.available_tools:validated_args = self.validate_arguments(tool_name, arguments)return tool_name, validated_argsexcept (json.JSONDecodeError, AttributeError):continue# 策略3: 基于关键词的启发式匹配for tool_name in self.available_tools:if tool_name.lower() in model_response.lower():arguments = self.extract_arguments_heuristic(model_response, tool_name)validated_args = self.validate_arguments(tool_name, arguments)return tool_name, validated_argsreturn None, Nonedef validate_arguments(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:"""验证参数是否符合工具的要求，并提供默认值"""if not arguments:arguments = {}if tool_name == "get_current_time" and "timezone" not in arguments:arguments["timezone"] = "UTC"return argumentsdef extract_arguments_heuristic(self, response: str, tool_name: str) -> Dict[str, Any]:"""使用启发式方法从响应中提取参数"""arguments = {}if tool_name == "get_current_time":timezone_patterns = [r'(asia/\w+|europe/\w+|america/\w+|australia/\w+|utc)',r'(上海|北京|纽约|伦敦|东京|悉尼)',r'(\bUTC\b|\bCST\b|\bEST\b|\bPST\b)']for pattern in timezone_patterns:match = re.search(pattern, response, re.IGNORECASE)if match:timezone = match.group(1)if timezone == "上海" or timezone == "北京":timezone = "Asia/Shanghai"elif timezone == "纽约":timezone = "America/New_York"elif timezone == "伦敦":timezone = "Europe/London"elif timezone == "东京":timezone = "Asia/Tokyo"elif timezone == "悉尼":timezone = "Australia/Sydney"arguments["timezone"] = timezonebreakreturn argumentsasync def chat_with_tools(self, user_query: str, model: str = "phi3:mini") -> str:"""与模型聊天，并处理可能的工具调用"""if not self._session:return "错误: MCP 会话未初始化"# 构建提示prompt = f"""用户查询: {user_query}你可用的工具: {json.dumps(list(self.tool_info.keys()), indent=2)}如果需要使用工具来回答问题，请以以下JSON格式回复:{{"name": "工具名称","arguments": {{"参数名": "参数值"}}}}例如，对于查询"现在上海是几点？"，你可以回复:{{"name": "get_current_time","arguments": {{"timezone": "Asia/Shanghai"}}}}如果不需要使用工具，请直接回复答案。"""try:# 调用Ollamaresponse = ollama.chat(model=model, messages=[{"role": "user", "content": prompt}])model_response = response['message']['content']print(f"模型初始响应: {model_response}")# 尝试提取工具调用信息tool_name, arguments = self.extract_tool_call(model_response)if tool_name and tool_name in self.available_tools:print(f"检测到工具调用: {tool_name}, 参数: {arguments}")# 调用工具try:tool_result = await self._session.call_tool(tool_name, arguments)tool_output = tool_result.contentprint(f"工具执行结果: {tool_output}")# 将工具结果发送回模型获取最终答案follow_up_prompt = f"""你之前决定调用工具 {tool_name} 来回答问题。工具执行结果: {tool_output}请基于这个结果生成对用户的最终回复。"""final_response = ollama.chat(model=model,messages=[{"role": "user", "content": follow_up_prompt}])return final_response['message']['content']except Exception as e:return f"调用工具时出错: {e}"else:return model_responseexcept Exception as e:return f"与模型通信时出错: {e}"async def main():# 替换为您的 time_server.py 的实际路径server_script_path = "/Users/mac/work/gitstudy/mcp-helloword/lzc-mcp/time_server.py"try:# 使用异步上下文管理器创建客户端并进行多次调用async with OllamaMCPClient(server_script_path) as client:# 测试查询 - 多次调用示例test_queries = ["现在上海是几点钟？","请问UTC时间现在是多少？","告诉我纽约的当前时间","你好，今天天气怎么样？"]for i, query in enumerate(test_queries, 1):print(f"\n{'=' * 50}")print(f"查询 #{i}: {query}")response = await client.chat_with_tools(query)print(f"最终回答: {response}")except Exception as e:print(f"程序执行出错: {e}")if __name__ == "__main__":asyncio.run(main())

智能决策流程：

提示工程：通过系统提示明确告知模型可用工具和调用格式
意图识别：模型分析用户查询，判断是否需要调用工具
JSON 解析：从模型响应中提取结构化工具调用指令
工具执行：通过 MCP 会话调用相应工具并返回结果

工作流程示例

时间查询场景

用户询问：“纽约现在是什么时间？”
大模型识别意图，生成工具调用指令：

{"action": "call_tool","tool_name": "get_current_time","arguments": {"timezone": "America/New_York"}
}

客户端调用 MCP 工具获取纽约当前时间
返回结果：“当前时间 (America/New_York) 是: 2024-01-15 10:30:45 EST”

无需工具场景

用户询问：“讲一个关于时间旅行的故事”
● 大模型识别无需调用工具，直接生成创意响应

附：上述客户端智能代理执行日志

处理查询: 现在几点了？(处理失败)

============================================================
处理查询: 现在几点了？
Connected to: <socket.socket fd=3, family=2, type=1, proto=0, laddr=('127.0.0.1', 57699), raddr=('127.0.0.1', 57696)>.
[09/16/25 21:27:36] INFO     Processing request of type            server.py:624ListToolsRequest                                   
已连接到时间服务器，可用工具: ['get_current_time', 'list_common_timezones']
大模型提示词: 你是一个AI助手，可以回答用户问题并决定是否需要调用时间服务。你可以使用的工具:- get_current_time: 获取指定时区的当前时间，参数: timezone (时区名称)- list_common_timezones: 获取常见时区列表，无参数调用格式:如果需要调用工具，请以以下JSON格式回复:{"action": "call_tool","tool_name": "工具名称","arguments": {参数键: 参数值}}如果不需要调用工具，请直接回复答案。当前可用工具: ['get_current_time', 'list_common_timezones']用户查询: 现在几点了？
询问大模型是否需要调用工具...
大模型初始响应: {"action": "call_tool","tool_name": "get_current_time","arguments": {"timezone": "Asia/Shanghai" // Assuming you're looking for the time in Shanghai, China.}
}
解析大模型响应时出错: Expecting ',' delimiter: line 5 column 37 (char 121)最终响应:
{"action": "call_tool","tool_name": "get_current_time","arguments": {"timezone": "Asia/Shanghai" // Assuming you're looking for the time in Shanghai, China.}
}
============================================================

处理查询: 纽约现在是什么时间？(处理成功)


============================================================
处理查询: 纽约现在是什么时间？
Connected to: <socket.socket fd=3, family=2, type=1, proto=0, laddr=('127.0.0.1', 57731), raddr=('127.0.0.1', 57696)>.
[09/16/25 21:27:41] INFO     Processing request of type            server.py:624ListToolsRequest                                   
已连接到时间服务器，可用工具: ['get_current_time', 'list_common_timezones']
大模型提示词: 你是一个AI助手，可以回答用户问题并决定是否需要调用时间服务。你可以使用的工具:- get_current_time: 获取指定时区的当前时间，参数: timezone (时区名称)- list_common_timezones: 获取常见时区列表，无参数调用格式:如果需要调用工具，请以以下JSON格式回复:{"action": "call_tool","tool_name": "工具名称","arguments": {参数键: 参数值}}如果不需要调用工具，请直接回复答案。当前可用工具: ['get_current_time', 'list_common_timezones']用户查询: 纽约现在是什么时间？
询问大模型是否需要调用工具...
大模型初始响应: 
{"action": "call_tool","tool_name": "get_current_time","arguments": {"timezone": "America/New_York"}
}根据JSON格式回答，你需要获取纽约当前的时间。为此，调用了“get_current_time”工具并指定了时区为“America/New_York”。现在请等待该API调用的结果后，你可以得到纽约当前的时间。大模型决定调用工具: get_current_time, 参数: {'timezone': 'America/New_York'}
[09/16/25 21:27:44] INFO     Processing request of type            server.py:624CallToolRequest                                    
工具调用结果: 当前时间 (America/New_York) 是: 2025-09-16 09:27:44 EDT最终响应:
当前时间 (America/New_York) 是: 2025-09-16 09:27:44 EDT
============================================================

处理查询: 给我列出一些常见的时区(处理失败)

============================================================
处理查询: 给我列出一些常见的时区
Connected to: <socket.socket fd=3, family=2, type=1, proto=0, laddr=('127.0.0.1', 57734), raddr=('127.0.0.1', 57696)>.
[09/16/25 21:27:45] INFO     Processing request of type            server.py:624ListToolsRequest                                   
已连接到时间服务器，可用工具: ['get_current_time', 'list_common_timezones']
大模型提示词: 你是一个AI助手，可以回答用户问题并决定是否需要调用时间服务。你可以使用的工具:- get_current_time: 获取指定时区的当前时间，参数: timezone (时区名称)- list_common_timezones: 获取常见时区列表，无参数调用格式:如果需要调用工具，请以以下JSON格式回复:{"action": "call_tool","tool_name": "工具名称","arguments": {参数键: 参数值}}如果不需要调用工具，请直接回复答案。当前可用工具: ['get_current_time', 'list_common_timezones']用户查询: 给我列出一些常见的时区
询问大模型是否需要调用工具...
大模型初始响应: {"action": "call_tool","tool_name": "list_common_timezones",
}根据常规时区的列表:
北美时区: America/New_York, America/Los_Angeles
澳大利亚时区: Australia/Sydney
东南亚时区: Asia/Shanghai, Asia/Kolkata
澳洲西部时区: Pacific/Noumea, Pacific/Guam
澳洲东部时区: Pacific/Auckland
澳洲北部时区: Pacific/Bougainville
澳洲南部时区: Pacific/Chuuk
澳洲西部时区: Pacific/Kiritimati
澳洲中部时区: Pacific/Pohnpei
澳洲南部时区: Pacific/Majuro
澳洲北部时区: Pacific/Galapagos
澳洲东部时区: Pacific/Enderbury
东盟时区: Asia/Bangkok
华北时区: Asia/Urumqi, Asia/Beijing
华南时区: Asia/Shanghai
中国东部时区: Asia/Chongqing
中国西部时区: Asia/Urumqi
香港时区: Asia/Hong Kong
台湾时区: Asia/Taipei, Asia/Macau
日本时区: Asia/Tokyo
东非时区: Africa/Johannesburg, Africa/Harare, Asia/Kuwait
中东时区: Asia/Riyadh, Europe/London
欧洲时区: Europe/Berlin, Europe/Istanbul
亚太时区: Asia/Dubai, Australia/Sydney
东南亚时区: Asia/Hong Kong, Indochina Time (ICT), Myanmar Standard Time (MST)
日内班岛时区: Asia/Ulan Bator
斯里兰大时区: America/Los_Angeles, Europe/London
澳洲：Asia/Sydney, Australia/Sydney
北美洲：America/New_York, America/Chicago对于一个非常规的时区，如北京时间(UTC+8)，我们可以直接写下回答:
北京时区是北方时间（UTC+8）。
解析大模型响应时出错: Illegal trailing comma before end of object: line 3 column 41 (char 69)最终响应:
{"action": "call_tool","tool_name": "list_common_timezones",
}根据常规时区的列表:
北美时区: America/New_York, America/Los_Angeles
澳大利亚时区: Australia/Sydney
东南亚时区: Asia/Shanghai, Asia/Kolkata
澳洲西部时区: Pacific/Noumea, Pacific/Guam
澳洲东部时区: Pacific/Auckland
澳洲北部时区: Pacific/Bougainville
澳洲南部时区: Pacific/Chuuk
澳洲西部时区: Pacific/Kiritimati
澳洲中部时区: Pacific/Pohnpei
澳洲南部时区: Pacific/Majuro
澳洲北部时区: Pacific/Galapagos
澳洲东部时区: Pacific/Enderbury
东盟时区: Asia/Bangkok
华北时区: Asia/Urumqi, Asia/Beijing
华南时区: Asia/Shanghai
中国东部时区: Asia/Chongqing
中国西部时区: Asia/Urumqi
香港时区: Asia/Hong Kong
台湾时区: Asia/Taipei, Asia/Macau
日本时区: Asia/Tokyo
东非时区: Africa/Johannesburg, Africa/Harare, Asia/Kuwait
中东时区: Asia/Riyadh, Europe/London
欧洲时区: Europe/Berlin, Europe/Istanbul
亚太时区: Asia/Dubai, Australia/Sydney
东南亚时区: Asia/Hong Kong, Indochina Time (ICT), Myanmar Standard Time (MST)
日内班岛时区: Asia/Ulan Bator
斯里兰大时区: America/Los_Angeles, Europe/London
澳洲：Asia/Sydney, Australia/Sydney
北美洲：America/New_York, America/Chicago对于一个非常规的时区，如北京时间(UTC+8)，我们可以直接写下回答:
北京时区是北方时间（UTC+8）。
============================================================

处理查询: Invalid/Timezone 现在的时间是多少？(处理失败)


============================================================
处理查询: Invalid/Timezone 现在的时间是多少？
Connected to: <socket.socket fd=3, family=2, type=1, proto=0, laddr=('127.0.0.1', 57766), raddr=('127.0.0.1', 57696)>.
[09/16/25 21:27:57] INFO     Processing request of type            server.py:624ListToolsRequest                                   
已连接到时间服务器，可用工具: ['get_current_time', 'list_common_timezones']
大模型提示词: 你是一个AI助手，可以回答用户问题并决定是否需要调用时间服务。你可以使用的工具:- get_current_time: 获取指定时区的当前时间，参数: timezone (时区名称)- list_common_timezones: 获取常见时区列表，无参数调用格式:如果需要调用工具，请以以下JSON格式回复:{"action": "call_tool","tool_name": "工具名称","arguments": {参数键: 参数值}}如果不需要调用工具，请直接回复答案。当前可用工具: ['get_current_time', 'list_common_timezones']用户查询: Invalid/Timezone 现在的时间是多少？
询问大模型是否需要调用工具...
大模型初始响应: 
{"action": "call_tool","tool_name": "get_current_time","arguments": {"timezone": "Asia/Shanghai" // Assuming the user wants to know the current time in Shanghai, China. You would need to replace this with the correct timezone based on the context of your question or if you are unsure about it ask for a list first using `list_common_timezones`.}
}
如果你不确定该哪个时区，可以先调用这个工具来获取常见的时区列表：{"action": "call_tool","tool_name": "list_common_timezones"
}
然后根据你的意愿或需求选择合适的时区，再使用 `get_current_time`。
解析大模型响应时出错: Expecting ',' delimiter: line 5 column 37 (char 121)最终响应:
{"action": "call_tool","tool_name": "get_current_time","arguments": {"timezone": "Asia/Shanghai" // Assuming the user wants to know the current time in Shanghai, China. You would need to replace this with the correct timezone based on the context of your question or if you are unsure about it ask for a list first using `list_common_timezones`.}
}
如果你不确定该哪个时区，可以先调用这个工具来获取常见的时区列表：{"action": "call_tool","tool_name": "list_common_timezones"
}
然后根据你的意愿或需求选择合适的时区，再使用 `get_current_time`。
============================================================

处理查询: 讲一个关于时间旅行的故事(处理成功)


============================================================
处理查询: 讲一个关于时间旅行的故事
Connected to: <socket.socket fd=3, family=2, type=1, proto=0, laddr=('127.0.0.1', 57770), raddr=('127.0.0.1', 57696)>.
[09/16/25 21:28:02] INFO     Processing request of type            server.py:624ListToolsRequest                                   
已连接到时间服务器，可用工具: ['get_current_time', 'list_common_timezones']
大模型提示词: 你是一个AI助手，可以回答用户问题并决定是否需要调用时间服务。你可以使用的工具:- get_current_time: 获取指定时区的当前时间，参数: timezone (时区名称)- list_common_timezones: 获取常见时区列表，无参数调用格式:如果需要调用工具，请以以下JSON格式回复:{"action": "call_tool","tool_name": "工具名称","arguments": {参数键: 参数值}}如果不需要调用工具，请直接回复答案。当前可用工具: ['get_current_time', 'list_common_timezones']用户查询: 讲一个关于时间旅行的故事
询问大模型是否需要调用工具...
大模型初始响应: 在黑洞漩曲中，李明和他的朋友小王发现了一个亚特僭的机器。这台机器能够将人们送回任何时代，无论是古老的三国时期还有未来。不久之后，李明和小王决定拓宽他们的视野，去遇见贵公主孙权。带着机器在古龙时代出行，两人以不可告別的身份和立场与其他同一时代的人物交流，试图改变历史中的不公正事件。在无数次的奇迹里找到了机会—-在一个有关科技知识传递和对未来发展影若的巨大突变时，他们被发现了并逃亡。不幸的是，他们无法通过这台机器抵御时间流变化，最终被各自的历史中心人物所强大的力量杀死了。然而，他们在其生命中的智慧和对时间旅行的深入了解，未来的科技家们会基于这些原则重新开发时间旅行机器，带来一个更加互联互信的世界。李明和小王的故事被记录在遗传编码中以后，成为了时间旅行理论和技术发展的一部分。尽管他们无法改变过去，但他们的决心和智慧对于科学界的进步与世界未来发展仍然有着永远的影响。最终响应:
在黑洞漩曲中，李明和他的朋友小王发现了一个亚特僭的机器。这台机器能够将人们送回任何时代，无论是古老的三国时期还有未来。不久之后，李明和小王决定拓宽他们的视野，去遇见贵公主孙权。带着机器在古龙时代出行，两人以不可告別的身份和立场与其他同一时代的人物交流，试图改变历史中的不公正事件。在无数次的奇迹里找到了机会—-在一个有关科技知识传递和对未来发展影若的巨大突变时，他们被发现了并逃亡。不幸的是，他们无法通过这台机器抵御时间流变化，最终被各自的历史中心人物所强大的力量杀死了。然而，他们在其生命中的智慧和对时间旅行的深入了解，未来的科技家们会基于这些原则重新开发时间旅行机器，带来一个更加互联互信的世界。李明和小王的故事被记录在遗传编码中以后，成为了时间旅行理论和技术发展的一部分。尽管他们无法改变过去，但他们的决心和智慧对于科学界的进步与世界未来发展仍然有着永远的影响。
============================================================