概述
在現代應用開發中,將大語言模型(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']用戶查詢: 講一個關于時間旅行的故事
詢問大模型是否需要調用工具...
大模型初始響應: 在黑洞漩曲中,李明和他的朋友小王發現了一個亞特僭的機器。這臺機器能夠將人們送回任何時代,無論是古老的三國時期還有未來。不久之后,李明和小王決定拓寬他們的視野,去遇見貴公主孫權。帶著機器在古龍時代出行,兩人以不可告別的身份和立場與其他同一時代的人物交流,試圖改變歷史中的不公正事件。在無數次的奇跡里找到了機會—-在一個有關科技知識傳遞和對未來發展影若的巨大突變時,他們被發現了并逃亡。不幸的是,他們無法通過這臺機器抵御時間流變化,最終被各自的歷史中心人物所強大的力量殺死了。然而,他們在其生命中的智慧和對時間旅行的深入了解,未來的科技家們會基于這些原則重新開發時間旅行機器,帶來一個更加互聯互信的世界。李明和小王的故事被記錄在遺傳編碼中以后,成為了時間旅行理論和技術發展的一部分。盡管他們無法改變過去,但他們的決心和智慧對于科學界的進步與世界未來發展仍然有著永遠的影響。最終響應:
在黑洞漩曲中,李明和他的朋友小王發現了一個亞特僭的機器。這臺機器能夠將人們送回任何時代,無論是古老的三國時期還有未來。不久之后,李明和小王決定拓寬他們的視野,去遇見貴公主孫權。帶著機器在古龍時代出行,兩人以不可告別的身份和立場與其他同一時代的人物交流,試圖改變歷史中的不公正事件。在無數次的奇跡里找到了機會—-在一個有關科技知識傳遞和對未來發展影若的巨大突變時,他們被發現了并逃亡。不幸的是,他們無法通過這臺機器抵御時間流變化,最終被各自的歷史中心人物所強大的力量殺死了。然而,他們在其生命中的智慧和對時間旅行的深入了解,未來的科技家們會基于這些原則重新開發時間旅行機器,帶來一個更加互聯互信的世界。李明和小王的故事被記錄在遺傳編碼中以后,成為了時間旅行理論和技術發展的一部分。盡管他們無法改變過去,但他們的決心和智慧對于科學界的進步與世界未來發展仍然有著永遠的影響。
============================================================
技術亮點
- 模塊化設計:服務與客戶端分離,符合微服務架構理念
- 協議標準化:使用 MCP 協議,確保工具調用的規范性和互操作性
- 本地化部署:使用 Ollama 和本地模型,保障數據隱私和響應速度
- 智能路由:大模型作為智能路由層,動態決定是否需要調用專業工具
- 錯誤恢復:完善的異常處理機制,確保系統穩定性
應用價值
這種架構模式具有廣泛的適用性:
- 企業級應用:可擴展為包含多種專業工具的企業智能助手
- 教育場景:演示如何將 LLM 與專業工具結合的教學案例
- 原型開發:快速構建領域特定智能應用的參考架構
- 研究平臺:探索工具使用和智能代理行為的實驗平臺
總結
本文介紹的系統展示了如何將大語言模型與專業工具服務通過 MCP 協議有機結合,創建出既能理解自然語言又能準確執行專業任務的智能代理。這種架構模式為構建下一代智能應用提供了可行路徑,既發揮了 LLM 的語言理解優勢,又保證了專業任務的執行準確性。