最近朋友圈和開發(fā)者群里經(jīng)常看到有人在刷“Agent 工作流”“MCP 服務(wù)”“技能包”“鉤子函數(shù)”這幾個詞GitHub 上相關(guān)的開源項目也是一個接一個地冒出來。尤其是想搞個人任務(wù)管理 Agent 的朋友幾乎都繞不開這套東西。不過說句實話網(wǎng)上很多資料要么只講概念不動手要么一上來就甩一堆框架看著高大上落地的時候處處是坑。所以這一期 GitHub 快報我想換個方式整理不單是盤點項目而是把 Agent 工作流、鉤子、技能、MCP 服務(wù)這四件事從頭到尾串起來講清楚它們之間到底是什么關(guān)系然后用一個“個人任務(wù)管理 Agent”的實際例子帶大家跑通一個最小可用的閉環(huán)。文章會包含完整代碼、配置文件、常見報錯和工程建議即使之前沒接觸過 MCP 或 Agent 概念也可以照著一步步做完。1. 這波 AI Agent 熱潮到底在聊什么先別急著寫代碼我們得先把幾個高頻詞捋清楚。因為很多人在 GitHub 上翻開源項目時經(jīng)?？吹?README 里同時出現(xiàn) Workflow、Hook、Skill、MCP完全分不清誰是誰也不知道自己的項目到底需要哪個。1.1 從“提示詞”到“工作流”最早大家和 ChatGPT 這類大模型聊天本質(zhì)上是在單輪對話里把需求講清楚模型直接給結(jié)果。但真實業(yè)務(wù)場景不可能這么簡單比如“幫我安排今天的任務(wù)還要考慮優(yōu)先級、截止時間、天氣通勤因素”這種需求如果只靠一段提示詞模型很容易漏掉條件回答也不穩(wěn)定。于是就有了 Agent 工作流。Agent 工作流指的不是某個單一模型調(diào)用而是把任務(wù)拆分成多個步驟每個步驟由一個或多個節(jié)點完成節(jié)點之間按照一定順序傳遞數(shù)據(jù)。比如一個典型的個人任務(wù)管理流程可以拆成收集任務(wù)信息。清洗和去重。調(diào)用日歷或待辦服務(wù)創(chuàng)建任務(wù)。根據(jù)優(yōu)先級和截止日期生成每日安排。把結(jié)果推送出去。這些步驟連接起來就是一條工作流。GitHub 上很多 Agent 框架比如 Dify、Coze、LangChain、n8n做的事情本質(zhì)上都是在幫我們描述和管理這種流程只是抽象層級不同。1.2 鉤子流程中的“攔截點”鉤子這個詞并不新鮮Git 有鉤子Redux 有中間件Web 開發(fā)里也有 Webhook。到了 Agent 工作流里鉤子依然是一種“在特定時機插入自定義邏輯”的機制。如果大家寫過鉤子函數(shù) C 語言示例或者用過 Git 的 pre-commit 鉤子應(yīng)該對這個概念不陌生。它的核心特點是某個事件發(fā)生前、發(fā)生后或者某個流程節(jié)點執(zhí)行前、執(zhí)行后系統(tǒng)會調(diào)用一個你預先注冊的函數(shù)。在 Agent 工作流里鉤子常被用來做這幾件事任務(wù)開始之前校驗輸入格式。大模型返回結(jié)果之后做敏感信息過濾。節(jié)點執(zhí)行失敗時觸發(fā)重試或告警。某個步驟完成后動態(tài)修改后續(xù)步驟的參數(shù)。舉個例子在個人任務(wù)管理 Agent 中用戶說“明天上午十點開會需要準備材料”工作流會先走到“意圖識別”節(jié)點然后走到“參數(shù)抽取”節(jié)點。如果我們希望在參數(shù)抽取完成后、創(chuàng)建任務(wù)之前檢查一下時間是否為工作日就可以在“創(chuàng)建任務(wù)”節(jié)點前掛一個鉤子。這樣邏輯更清晰不需要把校驗代碼寫死在業(yè)務(wù)節(jié)點內(nèi)部。1.3 技能讓 Agent 擁有“專項能力”技能Skill這個概念可以理解為一組預先封裝好的“能力包”。比如 ComfyUI 的技能包它把圖像生成所需的模型加載、采樣器配置、輸出格式都封裝起來用戶不需要關(guān)心底層細節(jié)直接拖一個技能節(jié)點到畫布上就能用。CTFHub 技能樹也是類似思路它把 Web 安全、逆向、密碼學等方向拆成可學習的技能點每一個技能點對應(yīng)一類工具和套路。放到 Agent 場景中技能是一個更上層的概念。一個技能通常包含能力描述告訴 Agent 這個技能能干什么。觸發(fā)條件什么情況下應(yīng)該調(diào)用它。輸入輸出定義需要什么參數(shù)會返回什么結(jié)果。底層實現(xiàn)具體調(diào)用哪個工具、哪個 API、哪段腳本。以個人任務(wù)管理 Agent 為例它可以具備“日程解析技能”“優(yōu)先級評估技能”“任務(wù)創(chuàng)建技能”“通勤時間計算技能”。每個技能對應(yīng)一個 Python 函數(shù)或一個 API 調(diào)用。Agent 的決策層負責根據(jù)用戶請求選擇合適的技能再串成一條執(zhí)行鏈。1.4 MCP 服務(wù)連接模型和外部世界的“標準插頭”MCP 全稱是 Model Context Protocol是一個開放協(xié)議目的是解決大模型與外部工具、數(shù)據(jù)源之間的連接標準化問題。在 MCP 出現(xiàn)之前每個 Agent 框架都有自己的工具調(diào)用規(guī)則接入一個新的待辦服務(wù)就要寫一套新的適配代碼。MCP 相當于定義了統(tǒng)一的“插頭規(guī)格”模型或 Agent 只要支持這個協(xié)議就能通過同一個標準去連接各種服務(wù)。一個 MCP 服務(wù)可以理解為“暴露給模型使用的一個工具集合”。它內(nèi)部包含若干工具Tools每個工具都聲明自己的輸入輸出結(jié)構(gòu)。Agent 可以通過 MCP 客戶端動態(tài)發(fā)現(xiàn)這些工具然后根據(jù)用戶需求決定調(diào)用哪些工具。GitHub 上已經(jīng)有很多現(xiàn)成的 MCP 服務(wù) demo比如數(shù)據(jù)庫 MCP、文件系統(tǒng) MCP、GitHub MCP 等。我們自己也可以開發(fā)一個私有的 MCP 服務(wù)把公司的待辦系統(tǒng)、日歷系統(tǒng)、知識庫接進去。這樣做的好處是業(yè)務(wù)邏輯只實現(xiàn)一次之后任何支持 MCP 的客戶端包括 Claude Desktop、各類 Agent 框架都能直接復用。2. 四者之間的關(guān)系用一張圖就能看懂很多教程喜歡把 Agent、工作流、鉤子、技能、MCP 分開講講完讀者還是懵的。下面我用文字描述一下它們?nèi)绾螀f(xié)作。先有一個 Agent 工作流它決定任務(wù)的整體流程。流程中有若干個節(jié)點每個節(jié)點可能執(zhí)行“調(diào)用大模型”“執(zhí)行代碼”“請求外部接口”等操作。鉤子附著在節(jié)點上負責在節(jié)點執(zhí)行前或執(zhí)行后插入自定義邏輯。比如記錄日志、動態(tài)修改請求參數(shù)、重試失敗節(jié)點。技能是比節(jié)點更高一層的封裝一個技能可能包含多個步驟和多個工具調(diào)用。工作流節(jié)點可以選擇某個技能來執(zhí)行具體任務(wù)。MCP 服務(wù)負責提供最底層的外部能力一個技能內(nèi)部可以調(diào)用一個或多個 MCP 工具而這些工具通過標準協(xié)議對外暴露。如果大家之前使用過 Dify 這類工作流平臺會發(fā)現(xiàn) Dify 中的“工具”節(jié)點實際上就可以對應(yīng)到 MCP 工具而“工作流”層面的條件分支、迭代節(jié)點配合“技能”插件機制正好覆蓋了四層結(jié)構(gòu)中的大部分。3. 環(huán)境準備開始動手前需要裝什么這一節(jié)我們了解一下后續(xù)演示要用到的環(huán)境。由于此類項目更新速度很快具體版本號不建議鎖死這里給出一個經(jīng)過驗證的常見組合大家根據(jù)實際網(wǎng)絡(luò)環(huán)境調(diào)整。3.1 運行環(huán)境操作系統(tǒng)macOS 或 Linux 或 Windows推薦使用 WSL2。Python 版本3.10 或更高。MCP SDK 和 Agent 框架對新版 Python 支持更好。Node.js可選部分 MCP 服務(wù)端示例基于 TypeScript我們這里統(tǒng)一用 Python。3.2 Python 依賴后續(xù)實戰(zhàn)環(huán)節(jié)會用到兩個核心庫一個是 MCP 官方 Python SDK我們可以通過 pip 安裝pip install mcp[cli]另一個是用于演示 Agent 工作流的輕量框架為了減少網(wǎng)絡(luò)和版本干擾這里我不依賴大型框架而是直接用 Python 的 asyncio 和 MCP SDK 手寫一個最小工作流引擎。這樣反而能讓大家看清楚內(nèi)部的執(zhí)行邏輯。如果安裝速度太慢可以臨時切換為內(nèi)部鏡像源例如pip install mcp[cli] -i https://pypi.tuna.tsinghua.edu.cn/simple安裝完成后可以驗證一下版本mcp --version python -c import mcp; print(mcp.__version__)3.3 個人任務(wù)管理服務(wù)的準備為了演示 MCP 服務(wù)我們不需要真的啟動一個復雜的日歷系統(tǒng)而是用 SQLite 本地數(shù)據(jù)庫來存儲任務(wù)這樣既輕量又能演示完整的增刪改查能力。SQLite 是 Python 標準庫自帶的模塊不需要額外安裝。數(shù)據(jù)庫文件就放在項目目錄下。如果后續(xù)需要接真實的 CalDAV 服務(wù)只需要在 MCP 服務(wù)內(nèi)部替換調(diào)用即可。3.4 項目結(jié)構(gòu)下面是我們即將創(chuàng)建的演示項目結(jié)構(gòu)task-agent/ ├── server/ │ ├── __init__.py │ └── task_mcp_server.py # MCP 服務(wù)端暴露任務(wù)管理工具 ├── workflow/ │ ├── __init__.py │ ├── engine.py # 迷你工作流引擎 │ ├── hooks.py # 鉤子注冊與觸發(fā) │ ├── skills.py # 技能定義與調(diào)度 │ └── client.py # MCP 客戶端連接服務(wù)端 ├── tasks.db # SQLite 數(shù)據(jù)庫運行時生成 └── requirements.txt這樣的結(jié)構(gòu)可以讓大家清晰地看到 MCP 服務(wù)、工作流、鉤子、技能分別落在哪些文件里而不是全堆在一個腳本里。4. 實踐從零構(gòu)建一個個人任務(wù)管理 Agent 工作流現(xiàn)在進入核心環(huán)節(jié)。這一節(jié)會分步驟實現(xiàn)一個“個人任務(wù)管理 Agent 工作流”整體流程如下用戶輸入一段自然語言比如“明天上午 10 點開會需要準備項目周報材料優(yōu)先級高”。工作流調(diào)用大模型接口做意圖識別和參數(shù)抽取。參數(shù)抽取完成后觸發(fā)一個鉤子校驗時間格式和截止日期。工作流調(diào)用“任務(wù)創(chuàng)建技能”技能內(nèi)部通過 MCP 客戶端調(diào)用本地 MCP 服務(wù)。MCP 服務(wù)把任務(wù)寫入 SQLite 數(shù)據(jù)庫。如果寫入成功再調(diào)用一個“日程提醒技能”計算提醒時間。最后輸出任務(wù) ID 和執(zhí)行結(jié)果。為了不依賴任何特定大模型廠商我們用一個 mock 函數(shù)來代替大模型調(diào)用。真實項目中只需要把這個函數(shù)替換為 OpenAI、通義千問、DeepSeek 等任意模型接口即可。4.1 定義任務(wù)數(shù)據(jù)模型首先在workflow目錄下新建一個models.py文件定義任務(wù)數(shù)據(jù)結(jié)構(gòu)和常量。# 文件路徑workflow/models.py from dataclasses import dataclass, field from typing import Optional dataclass class Task: title: str description: str priority: str medium # low / medium / high due_time: str remind_minutes: int 10 task_id: Optional[int] None def to_dict(self): return { task_id: self.task_id, title: self.title, description: self.description, priority: self.priority, due_time: self.due_time, remind_minutes: self.remind_minutes, }4.2 編寫 MCP 服務(wù)端下面這個文件是 MCP 服務(wù)端代碼它暴露了三個工具create_task、list_tasks、delete_task。使用 FastMCP 這個高級封裝可以大大減少樣板代碼。# 文件路徑server/task_mcp_server.py 一個最小的任務(wù)管理 MCP 服務(wù)端。 通過 FastMCP 封裝 SQLite 的增刪改查能力。 import sqlite3 import uuid from typing import List, Dict, Any from mcp.server.fastmcp import FastMCP mcp FastMCP(task-manager) DB_PATH tasks.db def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_conn() conn.execute( CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, title TEXT NOT NULL, description TEXT, priority TEXT DEFAULT medium, due_time TEXT, remind_minutes INTEGER DEFAULT 10, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() mcp.tool() def create_task( title: str, description: str , priority: str medium, due_time: str , remind_minutes: int 10, ) - Dict[str, Any]: 創(chuàng)建一條新的任務(wù)記錄返回任務(wù)ID和保存結(jié)果。 conn get_conn() task_id str(uuid.uuid4())[:8] conn.execute( INSERT INTO tasks (id, title, description, priority, due_time, remind_minutes) VALUES (?, ?, ?, ?, ?, ?) , (task_id, title, description, priority, due_time, remind_minutes), ) conn.commit() conn.close() return {task_id: task_id, status: success, title: title} mcp.tool() def list_tasks() - List[Dict[str, Any]]: 查詢當前全部任務(wù)列表。 conn get_conn() rows conn.execute(SELECT * FROM tasks ORDER BY created_at DESC).fetchall() conn.close() return [dict(row) for row in rows] mcp.tool() def delete_task(task_id: str) - Dict[str, Any]: 根據(jù)任務(wù)ID刪除一條任務(wù)。 conn get_conn() cursor conn.execute(DELETE FROM tasks WHERE id ?, (task_id,)) conn.commit() deleted cursor.rowcount conn.close() if deleted 0: return {status: error, message: 任務(wù)不存在} return {status: success, message: f已刪除任務(wù) {task_id}} if __name__ __main__: init_db() mcp.run(transportstdio)說明FastMCP 的mcp.tool()裝飾器可以把普通函數(shù)自動暴露為工具函數(shù)簽名會轉(zhuǎn)換成工具的 JSON Schema。這里使用transportstdio表示客戶端和服務(wù)端通過標準輸入輸出通信這種方式在本地開發(fā)中最方便。init_db()會在服務(wù)啟動前建好數(shù)據(jù)庫表避免首次調(diào)用時報錯。4.3 編寫 MCP 客戶端和工作流引擎接下來是工作流側(cè)。我們先實現(xiàn)一個非常輕量的工作流引擎然后用它來串聯(lián)整個任務(wù)管理流程。# 文件路徑workflow/engine.py 一個極簡的 Agent 工作流引擎。 核心思路 - 工作流由多個節(jié)點組成每個節(jié)點是一個 async 函數(shù)。 - 節(jié)點之間通過 context 字典共享數(shù)據(jù)。 - 每個節(jié)點可以聲明 before_hook 和 after_hook。 import asyncio import traceback from typing import Callable, Dict, Any class WorkflowNode: def __init__( self, name: str, handler: Callable[[Dict[str, Any]], Dict[str, Any]], before_hooksNone, after_hooksNone, ): self.name name self.handler handler self.before_hooks before_hooks or [] self.after_hooks after_hooks or [] async def run(self, context: Dict[str, Any]): # 執(zhí)行前鉤子 for hook in self.before_hooks: await hook(context, self.name, before) # 執(zhí)行主邏輯 result await self.handler(context) context[self.name] result # 執(zhí)行后鉤子 for hook in self.after_hooks: await hook(context, self.name, after) return result class Workflow: def __init__(self, name: str): self.name name self.nodes [] def add_node(self, node: WorkflowNode): self.nodes.append(node) return self async def run(self, initial_context: Dict[str, Any]): context initial_context.copy() for node in self.nodes: try: await node.run(context) except Exception as e: # 這里可以接入失敗重試或告警鉤子 print(f[{node.name}] 執(zhí)行失敗: {e}) traceback.print_exc() context[error] str(e) break return context這段代碼非常簡單但已經(jīng)具備了一個工作流引擎的核心按順序執(zhí)行節(jié)點、節(jié)點間通過 context 傳值、支持鉤子。真實框架會做得更復雜比如會有條件分支、循環(huán)節(jié)點、并行執(zhí)行但我們目前不需要。4.4 實現(xiàn)鉤子函數(shù)根據(jù)前面說的鉤子的作用是“在節(jié)點執(zhí)行前或執(zhí)行后插入邏輯”。下面我們寫一個鉤子模塊。# 文件路徑workflow/hooks.py 鉤子函數(shù)定義。 這里的鉤子是工作流節(jié)點范圍內(nèi)的鉤子。 import json from datetime import datetime async def validate_task_params_hook(context, node_name, stage): 在“創(chuàng)建任務(wù)”節(jié)點執(zhí)行前校驗參數(shù)是否合法。 if stage ! before: return parsed context.get(parsed_params, {}) title parsed.get(title, ).strip() if not title: raise ValueError(任務(wù)標題不能為空) due_time parsed.get(due_time, ) if due_time: try: datetime.fromisoformat(due_time) except ValueError: raise ValueError(f時間格式不合法: {due_time}請使用 ISO 格式例如 2025-01-01T10:00:00) print(f[hook] 參數(shù)校驗通過: {title}) async def log_node_result_hook(context, node_name, stage): 記錄節(jié)點執(zhí)行結(jié)果的鉤子。 if stage after and node_name in context: data context[node_name] # 只打印關(guān)鍵信息防止日志過大 summary data if isinstance(data, str) else str(data)[:200] print(f[hook] {node_name} 執(zhí)行完成結(jié)果摘要: {summary}) async def sanitize_output_hook(context, node_name, stage): 在“創(chuàng)建任務(wù)”節(jié)點執(zhí)行后對輸出做一次脫敏處理。 if stage ! after: return if node_name create_task and context.get(node_name): # 如果輸出中包含 error 信息這里可以決定是否屏蔽敏感字段 output context[node_name] if isinstance(output, dict) and status in output: context[node_name] { status: output[status], task_id: output.get(task_id), message: 任務(wù)處理完成, }這三個鉤子分別演示了三種典型用途輸入校驗。日志記錄。輸出后處理。如果大家以后接的是真實大模型可以在“生成回復”節(jié)點后加一個脫敏鉤子避免任務(wù)描述中的敏感信息直接暴露給用戶。4.5 實現(xiàn)技能調(diào)度技能不是某個具體函數(shù)而是一個“能力單元”的描述。下面用一個簡單的字典來定義技能元信息并實現(xiàn)一個最基礎(chǔ)的調(diào)度器。# 文件路徑workflow/skills.py 技能定義與調(diào)度。 一個技能包含 - name: 技能名稱 - description: 技能描述 - input_schema: 輸入?yún)?shù)說明 - handler: 執(zhí)行函數(shù)可以調(diào)用 MCP 工具 import json from typing import Callable, Dict, Any SKILL_REGISTRY: Dict[str, Dict[str, Any]] {} def register_skill(name: str, description: str, input_schema: Dict[str, Any]): def decorator(func: Callable[[Dict[str, Any]], Any]): SKILL_REGISTRY[name] { name: name, description: description, input_schema: input_schema, handler: func, } return func return decorator async def execute_skill(skill_name: str, params: Dict[str, Any]) - Any: 根據(jù)技能名稱找到對應(yīng)的 handler 并執(zhí)行。 if skill_name not in SKILL_REGISTRY: raise ValueError(f未知技能: {skill_name}) skill SKILL_REGISTRY[skill_name] return await skill[handler](params)這樣定義的好處是新增加一個技能只需要寫一個 async 函數(shù)并加上register_skill裝飾器即可不需要修改工作流主邏輯。4.6 編寫 Agent 工作流主流程下面我們把 MCP 客戶端、解析函數(shù)、技能調(diào)度和工作流引擎全部串起來。# 文件路徑workflow/client.py MCP 客戶端用于連接本地 MCP 服務(wù)。 import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class TaskMCPClient: def __init__(self, server_script: str): self.server_script server_script self.session None self._process None async def connect(self): server_params StdioServerParameters( commandpython, args[self.server_script], ) self._stack asyncio.Stack() self._process await self._stack.enter_async_context(stdio_client(server_params)) self._session await self._stack.enter_async_context(ClientSession(self._process[0], self._process[1])) await self._session.initialize() print([MCP 客戶端] 已連接到 task-manager 服務(wù)) async def call_tool(self, tool_name: str, arguments: dict): if not self._session: raise RuntimeError(MCP 客戶端尚未連接) result await self._session.call_tool(tool_name, arguments) # FastMCP 返回的內(nèi)容是一個列表其中每個元素有 text 字段 text for content in result.content: if hasattr(content, text): text content.text import json return json.loads(text) if text else {} async def close(self): if self._stack: await self._stack.aclose()注意上面代碼中asyncio.Stack()并不是 Python 標準用法實際上用于管理異步上下文的是AsyncExitStack正確的寫法如下from contextlib import AsyncExitStack class TaskMCPClient: def __init__(self, server_script: str): self.server_script server_script self.session None self._stack AsyncExitStack() async def connect(self): server_params StdioServerParameters( commandpython, args[self.server_script], ) self._stdio await self._stack.enter_async_context(stdio_client(server_params)) self._session await self._stack.enter_async_context(ClientSession(self._stdio[0], self._stdio[1])) await self._session.initialize() print([MCP 客戶端] 已連接到 task-manager 服務(wù)) async def call_tool(self, tool_name: str, arguments: dict): if not self._session: raise RuntimeError(MCP 客戶端尚未連接) result await self._session.call_tool(tool_name, arguments) text result.content[0].text return json.loads(text) async def close(self): await self._stack.aclose()下面定義主工作流腳本文件路徑可以命名為workflow/run_agent.py。# 文件路徑workflow/run_agent.py 個人任務(wù)管理 Agent 主流程。 示例輸入 明天上午 10 點開會需要準備項目周報材料優(yōu)先級高 import asyncio import json import os from datetime import datetime, timedelta from engine import Workflow, WorkflowNode from hooks import validate_task_params_hook, log_node_result_hook, sanitize_output_hook from skills import register_skill, execute_skill from client import TaskMCPClient # 模擬大模型解析函數(shù)真實項目中可以換成 LLM API 調(diào)用 async def mock_llm_parse(user_input: str) - dict: 模擬把用戶輸入解析成結(jié)構(gòu)化任務(wù)參數(shù)。 text user_input.lower() priority medium if 高 in user_input or urgent in text or high in text: priority high elif 低 in user_input or low in text: priority low due_time if 明天 in user_input: tomorrow datetime.now() timedelta(days1) if 上午 in user_input: due_time tomorrow.replace(hour10, minute0, second0, microsecond0).isoformat() else: due_time tomorrow.replace(hour18, minute0, second0, microsecond0).isoformat() elif 今天 in user_input: today datetime.now() if 下午 in user_input: due_time today.replace(hour15, minute0, second0, microsecond0).isoformat() else: due_time today.replace(hour12, minute0, second0, microsecond0).isoformat() # 簡單提取標題這里只做演示 title user_input.replace(優(yōu)先級高, ).replace(優(yōu)先級低, ).strip() if len(title) 20: title title[:20] ... return { title: title, description: user_input, priority: priority, due_time: due_time, remind_minutes: 30 if priority high else 10, } # 技能1任務(wù)創(chuàng)建技能 register_skill( namecreate_task_skill, description創(chuàng)建一條新的待辦任務(wù), input_schema{ type: object, properties: { title: {type: string}, description: {type: string}, priority: {type: string}, due_time: {type: string}, remind_minutes: {type: integer}, }, }, ) async def create_task_skill(params: dict): mcp TaskMCPClient(os.path.join(os.path.dirname(__file__), .., server, task_mcp_server.py)) await mcp.connect() try: result await mcp.call_tool(create_task, params) return result finally: await mcp.close() # 技能2任務(wù)查詢技能 register_skill( namelist_tasks_skill, description查看當前所有任務(wù), input_schema{type: object, properties: {}}, ) async def list_tasks_skill(params: dict): mcp TaskMCPClient(os.path.join(os.path.dirname(__file__), .., server, task_mcp_server.py)) await mcp.connect() try: result await mcp.call_tool(list_tasks, params) return result finally: await mcp.close() # 工作流節(jié)點處理函數(shù) async def parse_input_node(context): user_input context[user_input] parsed await mock_llm_parse(user_input) context[parsed_params] parsed return parsed async def create_task_node(context): params context[parsed_params] result await execute_skill(create_task_skill, params) context[task_result] result return result async def list_tasks_node(context): result await execute_skill(list_tasks_skill, {}) context[task_list] result return result async def generate_reply_node(context): task_result context.get(task_result, {}) task_list context.get(task_list, []) if task_result: if task_result.get(status) success: lines [ f任務(wù)創(chuàng)建成功。, f任務(wù) ID{task_result.get(task_id)}, f當前任務(wù)數(shù)量{len(task_list) if isinstance(task_list, list) else 0}, ] return \n.join(lines) return 任務(wù)創(chuàng)建失敗請檢查參數(shù)。 return 暫時沒有可執(zhí)行的任務(wù)操作。 async def main(): user_input 明天上午 10 點開會需要準備項目周報材料優(yōu)先級高 # 構(gòu)建工作流 wf Workflow(namepersonal-task-agent) wf.add_node(WorkflowNode( nameparse_input, handlerparse_input_node, after_hooks[log_node_result_hook], )) wf.add_node(WorkflowNode( namecreate_task, handlercreate_task_node, before_hooks[validate_task_params_hook, log_node_result_hook], after_hooks[log_node_result_hook, sanitize_output_hook], )) wf.add_node(WorkflowNode( namelist_tasks, handlerlist_tasks_node, after_hooks[log_node_result_hook], )) wf.add_node(WorkflowNode( namegenerate_reply, handlergenerate_reply_node, after_hooks[log_node_result_hook], )) # 執(zhí)行工作流 context await wf.run({user_input: user_input}) print(\n 最終回復 ) print(context.get(generate_reply, 無輸出)) if __name__ __main__: asyncio.run(main())這里需要提醒一下以上代碼是演示用的真實項目中的技能 handler 不應(yīng)該每次調(diào)用都重新 connect MCP 客戶端而應(yīng)該在啟動時復用同一個會話。我們這樣寫是為了讓示例足夠簡單大家理解思路即可。4.7 運行與結(jié)果說明在項目根目錄執(zhí)行cd task-agent python workflow/run_agent.py預期輸出類似于[hook] parse_input 執(zhí)行完成結(jié)果摘要: {title: 明天上午 10 點開會需要準備項目周報材料優(yōu)先級高, ...} [hook] 參數(shù)校驗通過: 明天上午 10 點開會需要準備項目周報材料優(yōu)先級高 [MCP 客戶端] 已連接到 task-manager 服務(wù) [hook] create_task 執(zhí)行完成結(jié)果摘要: {status: success, task_id: a1b2c3d4, title: 明天上午 10 點開會。} [MCP 客戶端] 已連接到 task-manager 服務(wù) [hook] list_tasks 執(zhí)行完成結(jié)果摘要: [{id: a1b2c3d4, title: 明天上午 10 點開會。, ...}] [hook] generate_reply 執(zhí)行完成結(jié)果摘要: 任務(wù)創(chuàng)建成功。任務(wù) IDa1b2c3d4當前任務(wù)數(shù)量1 最終回復 任務(wù)創(chuàng)建成功。 任務(wù) IDa1b2c3d4 當前任務(wù)數(shù)量1此時可以查看本地tasks.db數(shù)據(jù)庫確認任務(wù)已經(jīng)寫入。也可以手動啟動 MCP 服務(wù)端然后用命令行工具測試其他工具方法。5. 常見問題與排查思路這一部分我會把實際使用過程中最常遇到的一批問題整理成表格方便大家快速定位。問題現(xiàn)象常見原因解決思路mcp: command not foundPython 腳本目錄未加入 PATH檢查 Python 安裝位置或通過python -m mcp運行MCP 客戶端連接超時服務(wù)端腳本路徑錯誤或 Python 環(huán)境不一致確認服務(wù)端腳本絕對路徑使用同一個虛擬環(huán)境ModuleNotFoundError: No module named mcp未安裝 MCP SDK 或虛擬環(huán)境未激活執(zhí)行pip install mcp[cli]激活對應(yīng)虛擬環(huán)境調(diào)用工具時返回{status: error}參數(shù)格式錯誤或任務(wù) ID 不存在先調(diào)用 list_tasks 確認 ID 是否存在再檢查參數(shù)類型鉤子函數(shù)拋出的異常導致工作流中斷鉤子中使用了未捕獲的 ValueError在工作流引擎中捕獲異常并進行處理或改用日志記錄而不是拋錯GitHub 下載依賴速度極慢網(wǎng)絡(luò)鏈路問題設(shè)置鏡像源、使用代理需遵守本地法規(guī)、或下載離線 wheel 包安裝AsyncExitStack使用后連接未釋放忘記調(diào)用await client.close()使用asyncio的上下文管理方式確保 finally 中釋放資源下面挑兩個高頻問題展開說。5.1 MCP 工具返回內(nèi)容如何解析使用 FastMCP 時工具返回值會被包裝成CallToolResult其中的content是一個列表。如果工具返回的是 JSON 字符串列表中元素的text字段就是序列化后的 JSON。解析方式如下result await session.call_tool(create_task, arguments) for item in result.content: if hasattr(item, text): data json.loads(item.text) print(data)如果不做 JSON 解析直接打印result會看到一堆對象內(nèi)存地址這不是 bug只是協(xié)議層的包裝。在自建客戶端時建議封裝一個call_tool方法統(tǒng)一解析規(guī)則。5.2 鉤子拋異常導致流程中斷怎么辦鉤子函數(shù)里面拋ValueError或RuntimeError如果不是自己手動捕獲會中斷整個工作流。這在校驗類鉤子里其實是預期行為如果參數(shù)不合法就不應(yīng)該繼續(xù)執(zhí)行后續(xù)節(jié)點。但如果是日志鉤子拋異常就不應(yīng)該影響主流程了。一個比較好的實踐是日志類鉤子內(nèi)部捕獲全部異常只打印而不拋出校驗類鉤子則正常拋出讓工作流引擎處理終止邏輯。在引擎層面我們前面的簡單實現(xiàn)里已經(jīng)用了 try-except所以不會導致整個進程崩潰。6. 工程化建議如何把 Demo 變成可維護的系統(tǒng)到這里我們已經(jīng)跑通了一個最小可用的 Agent 工作流。但如果要在真實團隊中使用還有幾個方面值得優(yōu)化。6.1 鉤子要分級管理不要把所有鉤子都掛在同一個節(jié)點上。建議給鉤子增加級別Debug 級只輸出日志不影響流程。業(yè)務(wù)級做輸入校驗、參數(shù)修正、權(quán)限判斷。系統(tǒng)級做重試、熔斷、限流。不同級位對應(yīng)不同異常策略。系統(tǒng)級鉤子如果失敗要能觸發(fā)告警業(yè)務(wù)級鉤子失敗時可以返回錯誤信息給用戶Debug 級鉤子即使失敗也不要讓用戶感知。6.2 技能需要注冊表和版本管理當技能數(shù)量變多以后建議把技能注冊表抽出成一個 JSON 文件或數(shù)據(jù)庫表而不是堆在 Python 裝飾器里。每個技能應(yīng)該包含版本號、維護人、依賴項。升級技能時要像微服務(wù)升級 API 一樣考慮兼容性。一個推薦的結(jié)構(gòu)是{ name: create_task_skill, version: 1.2.0, description: 創(chuàng)建任務(wù)并寫入本地數(shù)據(jù)庫, inputs: { title: string, due_time: string(optional) }, outputs: { task_id: string, status: string }, runtime: python3.10 }這樣后續(xù)做權(quán)限控制、灰度發(fā)布、成本統(tǒng)計都會容易很多。6.3 MCP 服務(wù)要區(qū)分“本地長駐”和“遠程調(diào)用”我們演示中每個技能都重新連接一次 MCP 服務(wù)這在真實系統(tǒng)里不可取。生產(chǎn)環(huán)境通常有兩種模式本地長駐模式Agent 進程啟動時創(chuàng)建 MCP 客戶端連接多個技能共享同一個 session。遠程服務(wù)模式MCP 服務(wù)以 HTTP/SSE 方式部署客戶端通過 URL 連接。如果服務(wù)部署在公網(wǎng)必須加上身份認證和傳輸加密否則任何人都可能通過你的 MCP 服務(wù)讀寫任務(wù)數(shù)據(jù)。這是非常重要的一條安全紅線。6.4 日志和可觀測性Agent 工作流比普通接口鏈路長得多一個請求可能經(jīng)過大模型、技能、MCP、數(shù)據(jù)庫多個環(huán)節(jié)。建議從第一天就埋點至少要記錄每個節(jié)點的開始時間、結(jié)束時間、耗時。每次大模型調(diào)用的輸入輸出 token 數(shù)和費用。每次 MCP 工具調(diào)用的入?yún)ⅰ⒊鰠?、錯誤碼。鉤子觸發(fā)記錄。這些數(shù)據(jù)既可以用于排查問題也可以用來做成本分析和流程優(yōu)化。6.5 大模型解析結(jié)果要做兜底使用大模型解析用戶輸入時輸出格式并不總是穩(wěn)定的。即使加了 JSON Schema 約束模型偶爾也會返回不合法 JSON。真實項目中需要增加一層“解析結(jié)果校驗”固定范圍是模型輸出必須能轉(zhuǎn)為合法 JSON且 title 字段非空。如果校驗失敗可以讓模型重新生成一次或者回退到規(guī)則解析。6.6 安全與權(quán)限如果 Agent 可以操作數(shù)據(jù)庫、發(fā)送郵件、調(diào)用支付接口權(quán)限控制就必須前置。建議采用最小權(quán)限原則MCP 服務(wù)只暴露當前業(yè)務(wù)需要的工具。工具參數(shù)要做白名單校驗不能把用戶輸入直接傳給數(shù)據(jù)庫。刪除類操作必須二次確認。以刪除任務(wù)為例MCP 服務(wù)端應(yīng)該要求調(diào)用方傳入一個confirm字段值為yes時才真正執(zhí)行刪除。7. 后續(xù)還可以在哪些方向繼續(xù)深入如果我們已經(jīng)完成了上面這套個人任務(wù)管理 Agent接下來可以考慮往以下幾個方向做擴展。第一個方向是接入真實大模型解析能力。把mock_llm_parse函數(shù)替換為實際的 LLM API 調(diào)用之后整個工作流就能理解更復雜的自然語言比如“每周一早上提醒我寫周報順手把上周的任務(wù)歸檔”。這背后需要大模型具備工具調(diào)用能力而 MCP 正好提供了工具發(fā)現(xiàn)和調(diào)用標準。第二個方向是把任務(wù)存儲從 SQLite 換成云端服務(wù)。比如接入 Notion API 或 CalDAV 協(xié)議MCP 服務(wù)端的實現(xiàn)只需要改底層工作流層完全不用動。這正好體現(xiàn)了 MCP 協(xié)議的收益接入成本被限制在服務(wù)端而不是每個 Agent 客戶端。第三個方向是增加定時觸發(fā)能力。個人任務(wù)管理場景里很多任務(wù)是周期性的比如“每天早上九點生成待辦清單”。我們可以用 APScheduler 或 GitHub Actions 的 schedule 定時任務(wù)來觸發(fā)工作流把生成的待辦推送到釘釘、飛書或郵件。第四個方向是給技能增加“重試和降級”策略。當某個技能依賴的外部服務(wù)不可用時工作流可以選擇走降級路徑比如用本地規(guī)則替代大模型解析或者使用緩存數(shù)據(jù)生成回復。這也是 Agent 系統(tǒng)上生產(chǎn)環(huán)境必須考慮的問題。整個鏈路走通以后我個人覺得最有價值的并不是某個框架或協(xié)議本身而是這種“把模型能力、工具能力和流程編制能力組合起來”的思維方式。GitHub 上項目更新很快今天我們用的 MCP SDK 可能過幾個月就會出新版本但分層和抽象的底層邏輯一直有效工作流負責編排鉤子負責干預技能負責封裝能力MCP 負責標準連接。把這四層邊界劃清楚后面換模型、換服務(wù)、加能力都會比想象中順利。寫到這這一期 GitHub 快報的核心內(nèi)容就整理完了。里面涉及的示例代碼如果對大家有幫助可以直接復制到本地跑一跑遇到版本差異或者接口變動優(yōu)先查一下對應(yīng) SDK 的官方文檔就好。動手改一改比只看文章理解深得多。