在實際 AI 應用開發中構建一個能理解用戶意圖、調用工具并完成復雜任務的智能體Agent是核心挑戰。傳統的 LangChain Agent 框架雖然提供了基礎范式但在工具擴展性、協議標準化和技能復用性上仍存在瓶頸。當我們將 LangChain Agent 與模型上下文協議Model Context Protocol, MCP以及標準化的技能Skills體系相結合時Agent 的能力邊界和工程效率將得到顯著躍升。這種集成不僅讓 Agent 能動態接入海量外部工具和數據源還能實現技能的模塊化開發與跨項目復用為基于 Claude、GPT 等大模型構建更強大、更穩定的 AI 應用提供了堅實的技術底座。本文旨在深入解析 LangChain Agent 接入 MCP 與 Skills 的技術原理并通過一個從環境搭建到生產級實踐的全流程示例展示如何利用這套技術棧全方位提升開發效率與應用能力。我們將從核心概念入手逐步完成一個能查詢數據庫、調用 API 并處理文件的智能體構建并深入探討其中的配置細節、常見陷阱及性能優化策略。1. 理解 MCP 與 SkillsAgent 能力擴展的基石在深入代碼之前必須厘清 MCP 和 Skills 這兩個核心概念它們共同構成了現代 Agent 擴展能力的協議層和模塊層。1.1 模型上下文協議MCP是什么模型上下文協議是一種開放協議用于在大語言模型LLM與外部工具、數據源之間建立標準化的通信橋梁。你可以將其理解為 LLM 世界的“USB 協議”或“驅動程序接口標準”。在沒有 MCP 之前每個工具都需要為不同的 Agent 框架如 LangChain、LlamaIndex編寫特定的適配器代碼導致重復勞動和兼容性問題。MCP 的核心價值在于解耦與標準化對模型/Agent 而言它只需實現 MCP 客戶端就能接入任何遵循 MCP 協議的服務端Server所提供的工具無需關心工具的具體實現。對工具開發者而言只需將工具包裝成一個 MCP 服務端就能被所有支持 MCP 的 Agent 框架使用極大地擴展了工具的受眾。一個典型的 MCP 服務端會通過標準接口向客戶端“公布”自己提供了哪些工具Tools、數據源Resources以及提示詞模板Prompts。例如一個數據庫 MCP 服務端可能公布一個“執行 SQL 查詢”的工具一個天氣 API 的 MCP 服務端可能公布一個“獲取城市天氣”的工具。1.2 Skills可復用的能力模塊Skills技能是比工具Tools更高一層的抽象。一個 Skill 通常是為了完成一個特定領域任務而打包的一組工具、提示詞、工作流程甚至小模型。如果說工具是“螺絲刀”那么技能就是“組裝家具的完整工具箱和說明書”。在 LangChain 的生態中Skills 強調可復用性和組合性。例如“數據分析”技能可能包含數據加載、清洗、可視化和報告生成等多個工具和預設提示詞。通過將 Skills 與 MCP 結合我們可以實現動態發現與加載Agent 在運行時可以通過 MCP 發現并加載遠端服務器上的 Skills。版本管理與共享Skills 可以像軟件包一樣進行版本管理并在團隊或社區內共享。熱插拔無需重啟 Agent 服務即可動態添加或移除 Skills實現能力的靈活伸縮。1.3 LangChain Agent 的工作范式LangChain Agent 的核心思想是“推理-執行”循環。Agent 內部有一個 LLM 作為“大腦”它根據用戶輸入和當前上下文決定下一步是直接回答還是調用某個工具。調用工具后工具的執行結果會返回給 LLMLLM 再據此決定后續動作直到任務完成或達到終止條件。傳統的 LangChain Agent 在工具管理上相對靜態通常需要在代碼中顯式定義并傳入一個工具列表。而接入 MCP 后Agent 的工具列表可以動態地從多個 MCP 服務端獲取實現了工具管理的“云原生”化。2. 環境準備與核心依賴配置為了構建一個接入 MCP 與 Skills 的 LangChain Agent我們需要搭建一個包含客戶端、服務端和技能庫的完整開發環境。2.1 基礎環境與 Python 包管理建議使用 Python 3.10 或更高版本并使用虛擬環境隔離依賴。# 創建并激活虛擬環境以 conda 為例 conda create -n langchain-mcp-agent python3.10 conda activate langchain-mcp-agent # 使用 pip 安裝核心依賴 pip install langchain langchain-community langchain-core2.2 安裝 MCP 相關 SDKMCP 的實現通常包含客戶端庫和服務端開發庫。我們將使用mcp這個 Python SDK它提供了開發 MCP 組件所需的核心功能。# 安裝 MCP SDK pip install mcp # 安裝 LangChain 與 MCP 的集成庫如果官方或社區有提供 # 例如一個可能的集成包請根據實際生態調整 pip install langchain-mcp-integration注意MCP 生態仍在快速發展中具體的集成庫名稱可能變化。關鍵在于找到或實現一個能將 MCP 服務端提供的工具轉換為 LangChainTool對象的適配器。2.3 安裝示例 Skills 與工具服務端為了進行演示我們需要一些實際的 MCP 服務端來提供工具。這里以兩個常見的服務端為例文件系統服務端提供讀取、寫入、列出文件等工具。SQLite 數據庫服務端提供執行 SQL 查詢的工具。我們可以從社區尋找或自己實現這些服務端。假設我們使用一個名為mcp-server-filesystem和mcp-server-sqlite的包。# 安裝示例 MCP 服務端假設的包名請替換為實際可用的包 pip install mcp-server-filesystem mcp-server-sqlite2.4 配置大模型訪問本文以 Anthropic 的 Claude 模型為例你需要準備相應的 API 密鑰。其他模型如 OpenAI GPT 的配置邏輯類似。# 安裝 Claude SDK pip install anthropic在項目根目錄創建.env文件來管理敏感配置# .env ANTHROPIC_API_KEYyour_anthropic_api_key_here在代碼中通過python-dotenv加載# config.py import os from dotenv import load_dotenv load_dotenv() ANTHROPIC_API_KEY os.getenv(ANTHROPIC_API_KEY) if not ANTHROPIC_API_KEY: raise ValueError(請在 .env 文件中設置 ANTHROPIC_API_KEY)3. 構建一個動態工具發現的 LangChain Agent本節將分步構建一個核心 Agent它能夠從本地運行的多個 MCP 服務端動態發現工具并利用 Claude 模型進行推理和調用。3.1 啟動并連接 MCP 服務端首先我們需要在后臺啟動 MCP 服務端進程。在實際部署中這些服務端可能以獨立進程、容器或遠程服務的形式存在。這里我們在同一臺機器上以子進程方式啟動它們。# mcp_servers.py import subprocess import time import signal import sys class MCPServerManager: def __init__(self): self.servers [] def start_file_server(self): 啟動文件系統 MCP 服務端 # 假設服務端通過命令 mcp-server-filesystem 啟動監聽 8001 端口 cmd [mcp-server-filesystem, --root, ./data, --port, 8001] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE) self.servers.append((filesystem, proc, 8001)) time.sleep(2) # 等待服務端啟動 print(f文件系統 MCP 服務端已啟動 (PID: {proc.pid})) return proc def start_sqlite_server(self, db_path./data/example.db): 啟動 SQLite MCP 服務端 # 假設服務端通過命令 mcp-server-sqlite 啟動監聽 8002 端口 cmd [mcp-server-sqlite, --db, db_path, --port, 8002] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE) self.servers.append((sqlite, proc, 8002)) time.sleep(2) print(fSQLite MCP 服務端已啟動 (PID: {proc.pid})) return proc def stop_all(self): 停止所有 MCP 服務端 for name, proc, _ in self.servers: print(f正在停止 {name} 服務端...) proc.terminate() proc.wait() self.servers.clear() # 使用上下文管理器確保資源清理 if __name__ __main__: manager MCPServerManager() try: manager.start_file_server() manager.start_sqlite_server() print(所有 MCP 服務端已就緒按 CtrlC 停止...) while True: time.sleep(1) except KeyboardInterrupt: manager.stop_all()3.2 實現 MCP 客戶端并轉換為 LangChain Tools這是最關鍵的一步我們需要編寫一個 MCP 客戶端連接到服務端獲取其提供的工具列表并將每個工具包裝成 LangChain 能識別的Tool對象。# mcp_client.py import asyncio from typing import List, Optional from langchain.tools import BaseTool from langchain_core.tools import Tool from mcp import ClientSession, StdioServerParameters from mcp.client import stdio class MCPToolFetcher: def __init__(self, server_name: str, server_params: StdioServerParameters): self.server_name server_name self.server_params server_params self.tools: List[Tool] [] async def connect_and_fetch_tools(self): 連接到 MCP 服務端并獲取工具列表 # 創建與 MCP 服務端的會話 async with stdio.stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化連接 await session.initialize() # 列出服務端提供的所有工具 response await session.list_tools() mcp_tools response.tools for mcp_tool in mcp_tools: # 為每個 MCP 工具創建一個 LangChain Tool 包裝器 langchain_tool Tool( namef{self.server_name}_{mcp_tool.name}, funcself._create_tool_func(session, mcp_tool), descriptionmcp_tool.description, ) self.tools.append(langchain_tool) return self.tools def _create_tool_func(self, session, mcp_tool): 創建一個能調用特定 MCP 工具的同步函數 # 注意LangChain Tool 的 func 是同步的但 MCP 調用是異步的。 # 我們需要在同步函數中運行異步代碼。這里使用 asyncio.run 簡化處理 # 在生產環境中需要考慮更優的異步集成方式。 async def async_tool_func(**kwargs): # 調用 MCP 工具的 execute 方法 result await session.call_tool(mcp_tool.name, argumentskwargs) # 返回工具執行結果的文本內容 return \n.join([c.text for c in result.content if c.type text]) def sync_wrapper(**kwargs): # 在新的事件循環中運行異步函數適用于簡單腳本 # 注意在已有事件循環的環境中如 FastAPI需要使用其他方式 return asyncio.run(async_tool_func(**kwargs)) return sync_wrapper # 工具獲取工具函數 def get_all_mcp_tools() - List[Tool]: 獲取所有已配置 MCP 服務端的工具 all_tools [] # 定義服務端連接參數 servers [ (filesystem, StdioServerParameters(commandmcp-server-filesystem, args[--root, ./data])), (sqlite, StdioServerParameters(commandmcp-server-sqlite, args[--db, ./data/example.db])), ] async def fetch_all(): for name, params in servers: fetcher MCPToolFetcher(name, params) tools await fetcher.connect_and_fetch_tools() all_tools.extend(tools) # 運行異步函數獲取所有工具 asyncio.run(fetch_all()) return all_tools3.3 創建 LangChain Agent 并集成動態工具現在我們可以使用獲取到的動態工具列表來初始化一個 LangChain Agent。這里使用 ReAct 代理類型它適合多步驟的工具調用場景。# agent_builder.py from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_anthropic import ChatAnthropic from config import ANTHROPIC_API_KEY from mcp_client import get_all_mcp_tools def build_mcp_agent(): # 1. 初始化 Claude 模型 llm ChatAnthropic( modelclaude-3-haiku-20240307, # 可根據需要選擇 sonnet, opus 等型號 temperature0, api_keyANTHROPIC_API_KEY ) # 2. 動態獲取所有 MCP 工具 print(正在從 MCP 服務端發現工具...) tools get_all_mcp_tools() print(f已發現 {len(tools)} 個工具: {[t.name for t in tools]}) # 3. 定義 ReAct 代理的提示詞模板 # 提示詞需要指導模型如何思考和使用工具 prompt PromptTemplate.from_template( 你是一個有幫助的AI助手可以訪問以下工具 {tools} 請使用以下格式回答 問題用戶提出的問題 思考你需要思考如何一步步解決問題。你可以使用工具也可以直接回答。 行動要使用的工具名稱必須是以下工具之一[{tool_names}] 行動輸入工具的輸入必須是一個有效的JSON字符串 觀察工具返回的結果 ... (這個 思考/行動/行動輸入/觀察 循環可以重復多次) 思考我現在知道了最終答案 最終答案對原始問題的最終回答 開始 問題{input} 思考{agent_scratchpad} ) # 4. 創建 ReAct 代理 agent create_react_agent(llm, tools, prompt) # 5. 創建代理執行器控制最大迭代次數以避免無限循環 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 開啟詳細日志便于調試 handle_parsing_errorsTrue, # 處理模型輸出解析錯誤 max_iterations10, # 限制最大思考步驟 early_stopping_methodgenerate, # 當模型決定不再使用工具時停止 ) return agent_executor4. 運行驗證與結果分析構建好 Agent 后我們需要準備測試數據運行幾個典型任務來驗證其能力。4.1 準備測試環境與數據首先創建必要的目錄和測試數據。# 創建數據目錄和示例文件 mkdir -p ./data echo 項目報告草案\n主要內容...\n待辦整理圖表 ./data/report.txt echo 會議記錄\n日期2024-05-27\n議題Agent架構評審 ./data/meeting.txt # 創建并初始化一個 SQLite 示例數據庫 sqlite3 ./data/example.db EOF CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT, department TEXT); INSERT INTO users (name, email, department) VALUES (張三, zhangsanexample.com, 工程部), (李四, lisiexample.com, 產品部), (王五, wangwuexample.com, 市場部); EOF4.2 執行綜合任務測試現在編寫一個測試腳本讓 Agent 執行一個結合了文件操作和數據庫查詢的復雜任務。# run_agent.py from agent_builder import build_mcp_agent import asyncio async def main(): # 構建 Agent agent build_mcp_agent() # 測試任務 1簡單的文件讀取 print(\n 測試任務 1讀取文件 ) result1 await agent.ainvoke({input: 請讀取 data 目錄下 report.txt 文件的內容并總結其要點。}) print(f任務1結果: {result1[output]}) # 測試任務 2數據庫查詢 print(\n 測試任務 2查詢數據庫 ) result2 await agent.ainvoke({input: 查詢 example.db 數據庫中 users 表里所有在工程部的員工姓名和郵箱。}) print(f任務2結果: {result2[output]}) # 測試任務 3多步驟組合任務 print(\n 測試任務 3組合任務 ) result3 await agent.ainvoke({ input: 1. 首先請列出 data 目錄下所有的 .txt 文件。 2. 然后讀取 meeting.txt 文件提取會議日期。 3. 最后去 example.db 數據庫的 users 表里查一下產品部有哪些人把結果和會議日期一起整理成一個簡短的摘要。 }) print(f任務3結果: {result3[output]}) if __name__ __main__: asyncio.run(main())運行此腳本你應該能看到類似以下的輸出具體內容因模型隨機性略有不同正在從 MCP 服務端發現工具... 已發現 4 個工具: [filesystem_read_file, filesystem_list_directory, sqlite_execute_query, sqlite_list_tables] 測試任務 1讀取文件 進入新的 Agent 執行鏈... 思考用戶要求讀取 report.txt 文件并總結要點。我需要使用文件讀取工具。 行動filesystem_read_file 行動輸入{path: ./data/report.txt} 觀察項目報告草案\n主要內容...\n待辦整理圖表 思考我已讀取文件內容。現在需要總結要點。內容顯示這是一個項目報告草案包含主要內容和待辦事項整理圖表。我可以直接總結。 最終答案該文件是一個項目報告草案主要內容已列出當前待辦事項是整理圖表。 任務1結果該文件是一個項目報告草案主要內容已列出當前待辦事項是整理圖表。 測試任務 2查詢數據庫 進入新的 Agent 執行鏈... 思考用戶需要查詢工程部員工。我需要使用 SQLite 查詢工具。 行動sqlite_execute_query 行動輸入{query: SELECT name, email FROM users WHERE department 工程部} 觀察[{name: 張三, email: zhangsanexample.com}] 思考查詢返回了結果。我可以直接給出答案。 最終答案工程部的員工是張三郵箱是 zhangsanexample.com。 任務2結果工程部的員工是張三郵箱是 zhangsanexample.com。從輸出中我們可以看到 Agent 成功完成了以下工作動態工具發現啟動時從兩個 MCP 服務端獲取了 4 個工具。正確工具選擇針對“讀文件”任務選擇了filesystem_read_file針對“查數據庫”任務選擇了sqlite_execute_query。參數構造能夠根據任務描述正確構造工具所需的輸入參數如文件路徑、SQL 語句。結果解析與總結能夠理解工具返回的原始數據文本行、JSON 數組并將其組織成自然語言回答。4.3 關鍵配置參數解析在構建 Agent 時有幾個關鍵參數直接影響其行為和性能參數所在位置含義與影響推薦值/建議max_iterationsAgentExecutor代理最大推理-執行循環次數。防止任務過于復雜導致無限循環。簡單任務 5-10復雜任務 15-20。需結合max_tokens考慮。handle_parsing_errorsAgentExecutor是否處理模型輸出格式解析錯誤。開啟后解析失敗會嘗試讓模型重試。建議始終設為True提高魯棒性。verboseAgentExecutor是否打印詳細的思考鏈Chain of Thought日志。開發調試時設為True生產環境設為False。temperatureChatAnthropic模型生成文本的隨機性。值越高輸出越多樣、越有創造性。Agent 工具調用場景建議設為0或0.1以保證工具選擇和參數生成的穩定性。modelChatAnthropic使用的 Claude 模型版本。claude-3-haiku速度最快成本最低claude-3-sonnet平衡claude-3-opus能力最強但最慢最貴。根據任務復雜度選擇。5. 常見問題排查與性能優化將 LangChain Agent 與 MCP、Skills 集成到生產環境時會遇到一系列工程化挑戰。以下是典型問題的排查路徑和優化建議。5.1 連接與工具發現失敗現象Agent 啟動時報錯無法連接到 MCP 服務端或工具列表為空。可能原因檢查方式解決方案MCP 服務端未啟動檢查對應端口如 8001, 8002是否在監聽 (netstat -an | grep 8001)。查看MCPServerManager日志是否有啟動錯誤。確保啟動命令正確依賴已安裝。檢查服務端二進制文件路徑。命令或參數錯誤檢查StdioServerParameters中的command和args是否與服務端程序匹配。使用絕對路徑指定命令或確保命令在系統 PATH 中。參考服務端文檔確認參數格式。權限問題檢查服務端是否有權限訪問指定目錄如./data或數據庫文件。調整目錄權限或使用服務端可訪問的路徑。協議版本不兼容查看 MCP 客戶端和服務端的版本。檢查初始化握手階段的錯誤信息。確保客戶端和服務端使用的mcpSDK 版本兼容。可嘗試升級到最新穩定版。5.2 工具調用錯誤或超時現象Agent 選擇了正確的工具但調用失敗或長時間無響應。問題現象常見原因檢查方式處理建議工具參數格式錯誤模型生成的 JSON 參數不符合工具要求。查看verbose日志中的“行動輸入”字段。手動用相同參數測試工具。在提示詞中更清晰地描述工具所需的參數格式。使用 Pydantic 模型對工具輸入進行校驗。工具執行內部錯誤MCP 服務端在處理請求時崩潰或返回錯誤。查看 MCP 服務端進程的標準錯誤輸出。檢查服務端日志。確保輸入數據如 SQL 語法、文件路徑有效。在工具包裝函數中添加更詳細的錯誤捕獲和日志。網絡或進程通信超時服務端響應慢或進程僵死。在工具調用代碼處添加超時設置。監控服務端資源占用CPU/內存。為異步調用設置asyncio.wait_for超時。優化服務端性能或對耗時工具單獨設置更長的超時。異步上下文沖突在已有事件循環如 FastAPI中同步調用工具導致錯誤。觀察是否報錯RuntimeError: This event loop is already running。避免在同步函數中直接使用asyncio.run。改用asyncio.create_task或在主異步上下文中調用工具。重構代碼使整個 Agent 調用鏈保持異步。5.3 Agent 邏輯錯誤與優化現象Agent 陷入循環、選擇錯誤工具、或生成無關內容。問題根因分析優化策略工具選擇不準1. 工具描述 (description) 不夠清晰。2. 提示詞未充分指導模型如何選擇工具。3. 工具過多模型混淆。1.優化工具描述用自然語言清晰說明工具功能、輸入輸出示例。例如將sqlite_execute_query描述改為“執行一條 SQL SELECT 查詢語句并返回結果集。輸入應為包含query鍵的 JSON 對象。”2.改進提示詞在PromptTemplate中加入工具選擇范例。3.工具分組/路由對工具進行分類先讓 Agent 選擇大類再選擇具體工具。無效迭代過多模型在已經獲得答案的情況下仍繼續嘗試使用工具。1.調整max_iterations根據任務復雜度設置合理上限。2.使用更好的停止條件AgentExecutor的early_stopping_method設為generate讓模型自己決定何時停止。3.優化思考鏈在提示詞中強調“當你認為已有足夠信息回答問題時可以直接給出最終答案”。處理復雜任務能力弱單一 ReAct 代理難以規劃冗長或多分支任務。1.升級 Agent 類型使用Plan-and-Execute或OpenAI Functions代理它們更擅長規劃。2.引入 LangGraph對于有狀態、多分支的工作流使用 LangGraph 來顯式定義狀態圖和節點邏輯。3.任務分解在上層設計一個“主控”Agent負責將復雜任務拆解為子任務再分發給負責具體工具的“子”Agent。5.4 生產環境部署建議在開發環境跑通后部署到生產環境還需考慮以下方面MCP 服務端部署不應以簡單的子進程方式運行。建議將每個 MCP 服務端部署為獨立的容器Docker或系統服務systemd并配置健康檢查、資源限制和自動重啟。連接管理與池化頻繁創建銷毀到 MCP 服務端的連接開銷大。應實現連接池讓多個 Agent 實例共享到同一服務端的穩定連接。安全性工具權限嚴格限制每個 MCP 服務端的權限。例如文件系統服務端只允許訪問特定沙箱目錄數據庫服務端使用只讀或最小權限賬戶。輸入驗證與清理對所有從模型傳遞給工具的參數進行嚴格的驗證和清理防止 SQL 注入、路徑遍歷等攻擊。API 密鑰管理使用安全的秘密管理服務如 Vault, AWS Secrets Manager存儲和輪換 API 密鑰切勿硬編碼在代碼或配置文件中。可觀測性結構化日志記錄每個工具調用的詳細信息工具名、輸入、輸出、耗時、狀態。鏈路追蹤為每個用戶會話分配唯一 ID并貫穿所有的 Agent 思考、工具調用步驟便于問題排查。監控指標監控 Agent 的請求量、響應時間、工具調用成功率、迭代次數分布等。性能與成本緩存對頻繁且結果不變的工具調用如讀取靜態配置添加緩存層。模型選擇根據任務類型選擇合適的模型。簡單的工具調用可用Haiku復雜規劃可用Sonnet或Opus。限制 Token 消耗設置max_tokens上限防止因異常導致生成過長內容而產生高費用。6. 擴展方向構建自定義 Skills 與高級工作流掌握了基礎集成后你可以向兩個方向深入一是創建自己的 Skills二是構建更復雜的智能工作流。6.1 開發自定義 MCP 服務端與 Skill一個 Skill 本質上是一個或多個相關工具的集合并可能附帶一些預設提示詞。開發自定義 Skill 的步驟如下定義工具接口明確 Skill 要提供哪些功能。實現 MCP 服務端使用mcpSDK 實現這些功能并遵循 MCP 協議暴露它們。打包與分發將服務端代碼和配置打包如 Docker 鏡像、Python 包方便部署和共享。以下是一個簡單的“天氣查詢” Skill 的服務端示例# weather_mcp_server.py from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import asyncio from some_weather_lib import get_weather # 假設有一個獲取天氣的庫 async def main(): # 創建 MCP 服務端 server Server(weather-skills) server.list_tools() async def handle_list_tools(): # 公布此服務端提供的工具 return [ { name: get_current_weather, description: 獲取指定城市的當前天氣情況。, inputSchema: { type: object, properties: { city: {type: string, description: 城市名稱例如北京} }, required: [city] } } ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_current_weather: city arguments.get(city) if not city: raise ValueError(缺少參數 city) # 調用實際的外部天氣 API weather_info await get_weather(city) return [ { type: text, text: f{city}的天氣{weather_info[condition]}溫度{weather_info[temp]}°C。 } ] else: raise ValueError(f未知工具: {name}) # 通過標準輸入輸出運行服務端 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_nameweather-skills, server_version0.1.0 ) ) if __name__ __main__: asyncio.run(main())6.2 使用 LangGraph 編排復雜 Agent 工作流當任務涉及多個 Agent 協作、狀態持久化或條件分支時LangChain 的基礎 AgentExecutor 會顯得力不從心。此時LangGraph 是更強大的選擇。它允許你將工作流定義為圖Graph其中節點可以是 Agent、工具或任何函數邊定義了執行流程。例如一個“數據報告生成”工作流可能包含以下節點規劃節點分析用戶請求拆解為“取數據”、“分析”、“生成圖表”、“撰寫報告”等子任務。數據查詢節點調用 SQL MCP 工具獲取數據。分析節點調用 Python 計算工具或另一個 LLM 進行數據分析。圖表生成節點調用圖表生成 API。報告匯編節點將數據、分析結果、圖表整合成最終報告。使用 LangGraph你可以清晰地定義這些節點的執行順序和條件分支例如如果數據為空則跳過分析節點并持久化整個工作流的狀態實現更穩健和可調試的復雜 Agent 系統。通過將 LangChain Agent、MCP 協議和 Skills 模塊相結合我們構建的智能體不再是一個封閉、僵化的系統而是一個能夠動態擴展、靈活組合的開放平臺。這種架構使得集成新工具、復用已有能力、以及構建復雜工作流變得前所未有的高效。從簡單的文件查詢到結合數據庫、API 和自定義邏輯的復雜任務Agent 都能通過統一的協議層進行調度和執行。在向生產環境邁進時務必關注安全性、可靠性和可觀測性通過連接池、權限控制、結構化日志和監控指標來保障系統的穩定運行。下一步你可以嘗試開發自己的專屬 Skills或者利用 LangGraph 來設計更精巧的多智能體協作流程從而解鎖 AI 應用開發的更大潛力。