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