企业官网建设流程全解析

最近朋友圈和开发者群里经常看到有人在刷“Agent 工作流”“MCP 服务”“技能包”“钩子函数”这几个词，GitHub 上相关的开源项目也是一个接一个地冒出来。尤其是想搞个人任务管理 Agent 的朋友，几乎都绕不开这套东西。不过说句实话，网上很多资料要么只讲概念不动手，要么一上来就甩一堆框架，看着高大上，落地的时候处处是坑。

所以这一期 GitHub 快报，我想换个方式整理：不单是盘点项目，而是把 Agent 工作流、钩子、技能、MCP 服务这四件事从头到尾串起来，讲清楚它们之间到底是什么关系，然后用一个“个人任务管理 Agent”的实际例子，带大家跑通一个最小可用的闭环。文章会包含完整代码、配置文件、常见报错和工程建议，即使之前没接触过 MCP 或 Agent 概念，也可以照着一步步做完。

1. 这波 AI Agent 热潮到底在聊什么

先别急着写代码，我们得先把几个高频词捋清楚。因为很多人在 GitHub 上翻开源项目时，经常看到 README 里同时出现 Workflow、Hook、Skill、MCP，完全分不清谁是谁，也不知道自己的项目到底需要哪个。

1.1 从“提示词”到“工作流”

最早大家和 ChatGPT 这类大模型聊天，本质上是在单轮对话里把需求讲清楚，模型直接给结果。但真实业务场景不可能这么简单，比如“帮我安排今天的任务，还要考虑优先级、截止时间、天气通勤因素”，这种需求如果只靠一段提示词，模型很容易漏掉条件，回答也不稳定。于是就有了 Agent 工作流。

Agent 工作流指的不是某个单一模型调用，而是把任务拆分成多个步骤，每个步骤由一个或多个节点完成，节点之间按照一定顺序传递数据。比如一个典型的个人任务管理流程可以拆成：

1. 收集任务信息。
2. 清洗和去重。
3. 调用日历或待办服务创建任务。
4. 根据优先级和截止日期生成每日安排。
5. 把结果推送出去。

这些步骤连接起来，就是一条工作流。GitHub 上很多 Agent 框架，比如 Dify、Coze、LangChain、n8n，做的事情本质上都是在帮我们描述和管理这种流程，只是抽象层级不同。

1.2 钩子：流程中的“拦截点”

钩子这个词并不新鲜，Git 有钩子，Redux 有中间件，Web 开发里也有 Webhook。到了 Agent 工作流里，钩子依然是一种“在特定时机插入自定义逻辑”的机制。

如果大家写过钩子函数 C 语言示例，或者用过 Git 的 pre-commit 钩子，应该对这个概念不陌生。它的核心特点是：某个事件发生前、发生后，或者某个流程节点执行前、执行后，系统会调用一个你预先注册的函数。在 Agent 工作流里，钩子常被用来做这几件事：

• 任务开始之前校验输入格式。
• 大模型返回结果之后做敏感信息过滤。
• 节点执行失败时触发重试或告警。
• 某个步骤完成后，动态修改后续步骤的参数。

举个例子，在个人任务管理 Agent 中，用户说“明天上午十点开会，需要准备材料”，工作流会先走到“意图识别”节点，然后走到“参数抽取”节点。如果我们希望在参数抽取完成后、创建任务之前，检查一下时间是否为工作日，就可以在“创建任务”节点前挂一个钩子。这样逻辑更清晰，不需要把校验代码写死在业务节点内部。

1.3 技能：让 Agent 拥有“专项能力”

技能（Skill）这个概念，可以理解为一组预先封装好的“能力包”。比如 ComfyUI 的技能包，它把图像生成所需的模型加载、采样器配置、输出格式都封装起来，用户不需要关心底层细节，直接拖一个技能节点到画布上就能用。CTFHub 技能树也是类似思路，它把 Web 安全、逆向、密码学等方向拆成可学习的技能点，每一个技能点对应一类工具和套路。

放到 Agent 场景中，技能是一个更上层的概念。一个技能通常包含：

• 能力描述：告诉 Agent 这个技能能干什么。
• 触发条件：什么情况下应该调用它。
• 输入输出定义：需要什么参数，会返回什么结果。
• 底层实现：具体调用哪个工具、哪个 API、哪段脚本。

以个人任务管理 Agent 为例，它可以具备“日程解析技能”“优先级评估技能”“任务创建技能”“通勤时间计算技能”。每个技能对应一个 Python 函数或一个 API 调用。Agent 的决策层负责根据用户请求选择合适的技能，再串成一条执行链。

1.4 MCP 服务：连接模型和外部世界的“标准插头”

MCP 全称是 Model Context Protocol，是一个开放协议，目的是解决大模型与外部工具、数据源之间的连接标准化问题。在 MCP 出现之前，每个 Agent 框架都有自己的工具调用规则，接入一个新的待办服务，就要写一套新的适配代码。MCP 相当于定义了统一的“插头规格”，模型或 Agent 只要支持这个协议，就能通过同一个标准去连接各种服务。

一个 MCP 服务可以理解为“暴露给模型使用的一个工具集合”。它内部包含若干工具（Tools），每个工具都声明自己的输入输出结构。Agent 可以通过 MCP 客户端动态发现这些工具，然后根据用户需求决定调用哪些工具。

GitHub 上已经有很多现成的 MCP 服务 demo，比如数据库 MCP、文件系统 MCP、GitHub MCP 等。我们自己也可以开发一个私有的 MCP 服务，把公司的待办系统、日历系统、知识库接进去。这样做的好处是：业务逻辑只实现一次，之后任何支持 MCP 的客户端（包括 Claude Desktop、各类 Agent 框架）都能直接复用。

2. 四者之间的关系，用一张图就能看懂

很多教程喜欢把 Agent、工作流、钩子、技能、MCP 分开讲，讲完读者还是懵的。下面我用文字描述一下它们如何协作。

先有一个 Agent 工作流，它决定任务的整体流程。流程中有若干个节点，每个节点可能执行“调用大模型”“执行代码”“请求外部接口”等操作。

钩子附着在节点上，负责在节点执行前或执行后插入自定义逻辑。比如记录日志、动态修改请求参数、重试失败节点。

技能是比节点更高一层的封装，一个技能可能包含多个步骤和多个工具调用。工作流节点可以选择某个技能来执行具体任务。

MCP 服务负责提供最底层的外部能力，一个技能内部可以调用一个或多个 MCP 工具，而这些工具通过标准协议对外暴露。

如果大家之前使用过 Dify 这类工作流平台，会发现 Dify 中的“工具”节点实际上就可以对应到 MCP 工具，而“工作流”层面的条件分支、迭代节点，配合“技能”插件机制，正好覆盖了四层结构中的大部分。

3. 环境准备：开始动手前需要装什么

这一节我们了解一下后续演示要用到的环境。由于此类项目更新速度很快，具体版本号不建议锁死，这里给出一个经过验证的常见组合，大家根据实际网络环境调整。

3.1 运行环境

• 操作系统：macOS 或 Linux 或 Windows（推荐使用 WSL2）。
• Python 版本：3.10 或更高。MCP SDK 和 Agent 框架对新版 Python 支持更好。
• Node.js：可选，部分 MCP 服务端示例基于 TypeScript，我们这里统一用 Python。

3.2 Python 依赖

后续实战环节会用到两个核心库，一个是 MCP 官方 Python SDK，我们可以通过 pip 安装：

pip install "mcp[cli]"

另一个是用于演示 Agent 工作流的轻量框架，为了减少网络和版本干扰，这里我不依赖大型框架，而是直接用 Python 的 asyncio 和 MCP SDK 手写一个最小工作流引擎。这样反而能让大家看清楚内部的执行逻辑。

如果安装速度太慢，可以临时切换为内部镜像源，例如：

pip install "mcp[cli]" -i https://pypi.tuna.tsinghua.edu.cn/simple

安装完成后，可以验证一下版本：

mcp --version python -c "import mcp; print(mcp.__version__)"

3.3 个人任务管理服务的准备

为了演示 MCP 服务，我们不需要真的启动一个复杂的日历系统，而是用 SQLite 本地数据库来存储任务，这样既轻量又能演示完整的增删改查能力。

SQLite 是 Python 标准库自带的模块，不需要额外安装。数据库文件就放在项目目录下。如果后续需要接真实的 CalDAV 服务，只需要在 MCP 服务内部替换调用即可。

3.4 项目结构

下面是我们即将创建的演示项目结构：

task-agent/ ├── server/ │ ├── __init__.py │ └── task_mcp_server.py # MCP 服务端，暴露任务管理工具 ├── workflow/ │ ├── __init__.py │ ├── engine.py # 迷你工作流引擎 │ ├── hooks.py # 钩子注册与触发 │ ├── skills.py # 技能定义与调度 │ └── client.py # MCP 客户端，连接服务端 ├── tasks.db # SQLite 数据库（运行时生成） └── requirements.txt

这样的结构可以让大家清晰地看到 MCP 服务、工作流、钩子、技能分别落在哪些文件里，而不是全堆在一个脚本里。

4. 实践：从零构建一个个人任务管理 Agent 工作流

现在进入核心环节。这一节会分步骤实现一个“个人任务管理 Agent 工作流”，整体流程如下：

1. 用户输入一段自然语言，比如“明天上午 10 点开会，需要准备项目周报材料，优先级高”。
2. 工作流调用大模型接口做意图识别和参数抽取。
3. 参数抽取完成后，触发一个钩子，校验时间格式和截止日期。
4. 工作流调用“任务创建技能”，技能内部通过 MCP 客户端调用本地 MCP 服务。
5. MCP 服务把任务写入 SQLite 数据库。
6. 如果写入成功，再调用一个“日程提醒技能”，计算提醒时间。
7. 最后输出任务 ID 和执行结果。

为了不依赖任何特定大模型厂商，我们用一个 mock 函数来代替大模型调用。真实项目中，只需要把这个函数替换为 OpenAI、通义千问、DeepSeek 等任意模型接口即可。

4.1 定义任务数据模型

首先在workflow目录下新建一个models.py文件，定义任务数据结构和常量。

# 文件路径：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 服务端

下面这个文件是 MCP 服务端代码，它暴露了三个工具：create_task、list_tasks、delete_task。使用 FastMCP 这个高级封装可以大大减少样板代码。

# 文件路径：server/task_mcp_server.py """ 一个最小的任务管理 MCP 服务端。 通过 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]: """创建一条新的任务记录，返回任务ID和保存结果。""" 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]]: """查询当前全部任务列表。""" 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]: """根据任务ID删除一条任务。""" 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": "任务不存在"} return {"status": "success", "message": f"已删除任务 {task_id}"} if __name__ == "__main__": init_db() mcp.run(transport="stdio")

说明：

• FastMCP 的mcp.tool()装饰器可以把普通函数自动暴露为工具，函数签名会转换成工具的 JSON Schema。
• 这里使用transport="stdio"，表示客户端和服务端通过标准输入输出通信，这种方式在本地开发中最方便。
• init_db()会在服务启动前建好数据库表，避免首次调用时报错。

4.3 编写 MCP 客户端和工作流引擎

接下来是工作流侧。我们先实现一个非常轻量的工作流引擎，然后用它来串联整个任务管理流程。

# 文件路径：workflow/engine.py """ 一个极简的 Agent 工作流引擎。 核心思路： - 工作流由多个节点组成，每个节点是一个 async 函数。 - 节点之间通过 context 字典共享数据。 - 每个节点可以声明 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_hooks=None, after_hooks=None, ): 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]): # 执行前钩子 for hook in self.before_hooks: await hook(context, self.name, "before") # 执行主逻辑 result = await self.handler(context) context[self.name] = result # 执行后钩子 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}] 执行失败: {e}") traceback.print_exc() context["error"] = str(e) break return context

这段代码非常简单，但已经具备了一个工作流引擎的核心：按顺序执行节点、节点间通过 context 传值、支持钩子。真实框架会做得更复杂，比如会有条件分支、循环节点、并行执行，但我们目前不需要。

4.4 实现钩子函数

根据前面说的，钩子的作用是“在节点执行前或执行后插入逻辑”。下面我们写一个钩子模块。

# 文件路径：workflow/hooks.py """ 钩子函数定义。 这里的钩子是工作流节点范围内的钩子。 """ import json from datetime import datetime async def validate_task_params_hook(context, node_name, stage): """ 在“创建任务”节点执行前，校验参数是否合法。 """ if stage != "before": return parsed = context.get("parsed_params", {}) title = parsed.get("title", "").strip() if not title: raise ValueError("任务标题不能为空") 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] 参数校验通过: {title}") async def log_node_result_hook(context, node_name, stage): """ 记录节点执行结果的钩子。 """ if stage == "after" and node_name in context: data = context[node_name] # 只打印关键信息，防止日志过大 summary = data if isinstance(data, str) else str(data)[:200] print(f"[hook] {node_name} 执行完成，结果摘要: {summary}") async def sanitize_output_hook(context, node_name, stage): """ 在“创建任务”节点执行后，对输出做一次脱敏处理。 """ 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": "任务处理完成", }

这三个钩子分别演示了三种典型用途：

1. 输入校验。
2. 日志记录。
3. 输出后处理。

如果大家以后接的是真实大模型，可以在“生成回复”节点后加一个脱敏钩子，避免任务描述中的敏感信息直接暴露给用户。

4.5 实现技能调度

技能不是某个具体函数，而是一个“能力单元”的描述。下面用一个简单的字典来定义技能元信息，并实现一个最基础的调度器。

# 文件路径：workflow/skills.py """ 技能定义与调度。 一个技能包含： - name: 技能名称 - description: 技能描述 - input_schema: 输入参数说明 - handler: 执行函数，可以调用 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: """根据技能名称，找到对应的 handler 并执行。""" if skill_name not in SKILL_REGISTRY: raise ValueError(f"未知技能: {skill_name}") skill = SKILL_REGISTRY[skill_name] return await skill["handler"](params)

这样定义的好处是：新增加一个技能，只需要写一个 async 函数并加上@register_skill装饰器即可，不需要修改工作流主逻辑。

4.6 编写 Agent 工作流主流程

下面我们把 MCP 客户端、解析函数、技能调度和工作流引擎全部串起来。

# 文件路径：workflow/client.py """ MCP 客户端，用于连接本地 MCP 服务。 """ 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( command="python", 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 服务") 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 返回的内容是一个列表，其中每个元素有 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( command="python", 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 服务") 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 """ 个人任务管理 Agent 主流程。 示例输入： "明天上午 10 点开会，需要准备项目周报材料，优先级高" """ 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 # 模拟大模型解析函数，真实项目中可以换成 LLM API 调用 async def mock_llm_parse(user_input: str) -> dict: """ 模拟把用户输入解析成结构化任务参数。 """ 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(days=1) if "上午" in user_input: due_time = tomorrow.replace(hour=10, minute=0, second=0, microsecond=0).isoformat() else: due_time = tomorrow.replace(hour=18, minute=0, second=0, microsecond=0).isoformat() elif "今天" in user_input: today = datetime.now() if "下午" in user_input: due_time = today.replace(hour=15, minute=0, second=0, microsecond=0).isoformat() else: due_time = today.replace(hour=12, minute=0, second=0, microsecond=0).isoformat() # 简单提取标题，这里只做演示 title = user_input.replace("优先级高", "").replace("优先级低", "").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：任务创建技能 @register_skill( name="create_task_skill", description="创建一条新的待办任务", 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：任务查询技能 @register_skill( name="list_tasks_skill", description="查看当前所有任务", 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() # 工作流节点处理函数 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"任务创建成功。", f"任务 ID：{task_result.get('task_id')}", f"当前任务数量：{len(task_list) if isinstance(task_list, list) else 0}", ] return "\n".join(lines) return "任务创建失败，请检查参数。" return "暂时没有可执行的任务操作。" async def main(): user_input = "明天上午 10 点开会，需要准备项目周报材料，优先级高" # 构建工作流 wf = Workflow(name="personal-task-agent") wf.add_node(WorkflowNode( name="parse_input", handler=parse_input_node, after_hooks=[log_node_result_hook], )) wf.add_node(WorkflowNode( name="create_task", handler=create_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( name="list_tasks", handler=list_tasks_node, after_hooks=[log_node_result_hook], )) wf.add_node(WorkflowNode( name="generate_reply", handler=generate_reply_node, after_hooks=[log_node_result_hook], )) # 执行工作流 context = await wf.run({"user_input": user_input}) print("\n===== 最终回复 =====") print(context.get("generate_reply", "无输出")) if __name__ == "__main__": asyncio.run(main())

这里需要提醒一下，以上代码是演示用的，真实项目中的技能 handler 不应该每次调用都重新 connect MCP 客户端，而应该在启动时复用同一个会话。我们这样写是为了让示例足够简单，大家理解思路即可。

4.7 运行与结果说明

在项目根目录执行：

cd task-agent python workflow/run_agent.py

预期输出类似于：

[hook] parse_input 执行完成，结果摘要: {'title': '明天上午 10 点开会，需要准备项目周报材料，优先级高', ...} [hook] 参数校验通过: 明天上午 10 点开会，需要准备项目周报材料，优先级高 [MCP 客户端] 已连接到 task-manager 服务 [hook] create_task 执行完成，结果摘要: {'status': 'success', 'task_id': 'a1b2c3d4', 'title': '明天上午 10 点开会。'} [MCP 客户端] 已连接到 task-manager 服务 [hook] list_tasks 执行完成，结果摘要: [{'id': 'a1b2c3d4', 'title': '明天上午 10 点开会。', ...}] [hook] generate_reply 执行完成，结果摘要: 任务创建成功。任务 ID：a1b2c3d4，当前任务数量：1 ===== 最终回复 ===== 任务创建成功。 任务 ID：a1b2c3d4 当前任务数量：1

此时可以查看本地tasks.db数据库，确认任务已经写入。也可以手动启动 MCP 服务端，然后用命令行工具测试其他工具方法。

5. 常见问题与排查思路

这一部分我会把实际使用过程中最常遇到的一批问题整理成表格，方便大家快速定位。

下面挑两个高频问题展开说。

5.1 MCP 工具返回内容如何解析

使用 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会看到一堆对象内存地址，这不是 bug，只是协议层的包装。在自建客户端时，建议封装一个call_tool方法，统一解析规则。

5.2 钩子抛异常导致流程中断怎么办

钩子函数里面抛ValueError或RuntimeError，如果不是自己手动捕获，会中断整个工作流。这在校验类钩子里其实是预期行为：如果参数不合法，就不应该继续执行后续节点。

但如果是日志钩子抛异常，就不应该影响主流程了。一个比较好的实践是：日志类钩子内部捕获全部异常，只打印而不抛出；校验类钩子则正常抛出，让工作流引擎处理终止逻辑。在引擎层面，我们前面的简单实现里已经用了 try-except，所以不会导致整个进程崩溃。

6. 工程化建议：如何把 Demo 变成可维护的系统

到这里，我们已经跑通了一个最小可用的 Agent 工作流。但如果要在真实团队中使用，还有几个方面值得优化。

6.1 钩子要分级管理

不要把所有钩子都挂在同一个节点上。建议给钩子增加级别：

• Debug 级：只输出日志，不影响流程。
• 业务级：做输入校验、参数修正、权限判断。
• 系统级：做重试、熔断、限流。

不同级位对应不同异常策略。系统级钩子如果失败，要能触发告警；业务级钩子失败时，可以返回错误信息给用户；Debug 级钩子即使失败也不要让用户感知。

6.2 技能需要注册表和版本管理

当技能数量变多以后，建议把技能注册表抽出成一个 JSON 文件或数据库表，而不是堆在 Python 装饰器里。每个技能应该包含版本号、维护人、依赖项。升级技能时，要像微服务升级 API 一样考虑兼容性。

一个推荐的结构是：

{ "name": "create_task_skill", "version": "1.2.0", "description": "创建任务并写入本地数据库", "inputs": { "title": "string", "due_time": "string(optional)" }, "outputs": { "task_id": "string", "status": "string" }, "runtime": "python3.10" }

这样后续做权限控制、灰度发布、成本统计都会容易很多。

6.3 MCP 服务要区分“本地长驻”和“远程调用”

我们演示中每个技能都重新连接一次 MCP 服务，这在真实系统里不可取。生产环境通常有两种模式：

• 本地长驻模式：Agent 进程启动时创建 MCP 客户端连接，多个技能共享同一个 session。
• 远程服务模式：MCP 服务以 HTTP/SSE 方式部署，客户端通过 URL 连接。

如果服务部署在公网，必须加上身份认证和传输加密，否则任何人都可能通过你的 MCP 服务读写任务数据。这是非常重要的一条安全红线。

6.4 日志和可观测性

Agent 工作流比普通接口链路长得多，一个请求可能经过大模型、技能、MCP、数据库多个环节。建议从第一天就埋点，至少要记录：

• 每个节点的开始时间、结束时间、耗时。
• 每次大模型调用的输入输出 token 数和费用。
• 每次 MCP 工具调用的入参、出参、错误码。
• 钩子触发记录。

这些数据既可以用于排查问题，也可以用来做成本分析和流程优化。

6.5 大模型解析结果要做兜底

使用大模型解析用户输入时，输出格式并不总是稳定的。即使加了 JSON Schema 约束，模型偶尔也会返回不合法 JSON。真实项目中，需要增加一层“解析结果校验”，固定范围是：模型输出必须能转为合法 JSON，且 title 字段非空。如果校验失败，可以让模型重新生成一次，或者回退到规则解析。

6.6 安全与权限

如果 Agent 可以操作数据库、发送邮件、调用支付接口，权限控制就必须前置。建议采用最小权限原则：

• MCP 服务只暴露当前业务需要的工具。
• 工具参数要做白名单校验，不能把用户输入直接传给数据库。
• 删除类操作必须二次确认。

以删除任务为例，MCP 服务端应该要求调用方传入一个confirm字段，值为yes时才真正执行删除。

7. 后续还可以在哪些方向继续深入

如果我们已经完成了上面这套个人任务管理 Agent，接下来可以考虑往以下几个方向做扩展。

第一个方向是接入真实大模型解析能力。把mock_llm_parse函数替换为实际的 LLM API 调用之后，整个工作流就能理解更复杂的自然语言，比如“每周一早上提醒我写周报，顺手把上周的任务归档”。这背后需要大模型具备工具调用能力，而 MCP 正好提供了工具发现和调用标准。

第二个方向是把任务存储从 SQLite 换成云端服务。比如接入 Notion API 或 CalDAV 协议，MCP 服务端的实现只需要改底层，工作流层完全不用动。这正好体现了 MCP 协议的收益：接入成本被限制在服务端，而不是每个 Agent 客户端。

第三个方向是增加定时触发能力。个人任务管理场景里，很多任务是周期性的，比如“每天早上九点生成待办清单”。我们可以用 APScheduler 或 GitHub Actions 的 schedule 定时任务来触发工作流，把生成的待办推送到钉钉、飞书或邮件。

第四个方向是给技能增加“重试和降级”策略。当某个技能依赖的外部服务不可用时，工作流可以选择走降级路径，比如用本地规则替代大模型解析，或者使用缓存数据生成回复。这也是 Agent 系统上生产环境必须考虑的问题。

整个链路走通以后，我个人觉得最有价值的并不是某个框架或协议本身，而是这种“把模型能力、工具能力和流程编制能力组合起来”的思维方式。GitHub 上项目更新很快，今天我们用的 MCP SDK 可能过几个月就会出新版本，但分层和抽象的底层逻辑一直有效：工作流负责编排，钩子负责干预，技能负责封装能力，MCP 负责标准连接。把这四层边界划清楚，后面换模型、换服务、加能力，都会比想象中顺利。

写到这，这一期 GitHub 快报的核心内容就整理完了。里面涉及的示例代码如果对大家有帮助，可以直接复制到本地跑一跑，遇到版本差异或者接口变动，优先查一下对应 SDK 的官方文档就好。动手改一改，比只看文章理解深得多。