Agent工作流、钩子、技能与MCP服务：从概念到工程实践

GitHub 快报第 392 期的热门内容里Agent 工作流、钩子、技能、MCP 服务是四个出现频率很高的方向。它们看起来是四个独立的关键词实际指向同一个趋势AI Agent 正在从一次对话式的 demo变成需要编排、扩展和统一接入外部工具的工程系统。很多人看到这些开源项目时会先收藏但真正要用起来还需要先搞清每个概念解决什么问题、项目里哪些文件值得读、自己动手复刻时从哪一步开始。这篇文章就以这期快报涉及的四类项目为背景先讲概念再讲项目评估方法然后给出一个可运行的最小工作流和一个 MCP 服务 Demo最后整理常见坑和可复用的检查清单。1. 先分清 Agent 工作流、钩子、技能、MCP 服务的分工1.1 Agent 工作流把一次对话变成一次任务执行Agent 工作流解决的核心问题是如何让模型在多个步骤之间保持一致的目标而不是单次问答。单次 LLM 调用擅长生成文本不擅长保证“先查数据、再判断、再执行、最后确认”的顺序。工作流把任务拆成一个有向的执行过程每个节点做一件确定的事。从技术角度看工作流由节点和连线组成。节点可以是 LLM 调用、工具调用、条件判断、循环、人工确认等类型。连线表示数据传递和执行顺序。一个典型的任务型工作流可能是这样的解析用户输入提取任务要素调用检索或查询工具取得上下文让 LLM 基于上下文生成执行方案必要时请求用户确认执行最终动作并返回结果。为什么要这样设计因为每一步的执行结果都需要被记录、验证和回滚。直接让 LLM 生成一大段答案出了问题很难定位是哪一段逻辑错了。拆成步骤后可以给每一步加超时、重试、日志和人工审核。这里容易误解的是工作流不等于写一串提示词。提示词只在单个节点内起作用工作流控制的是节点之间的流转关系。提示词负责“生成”工作流负责“编排”。1.2 钩子在固定节点插入自定义逻辑钩子是一个提前留好的插槽。框架或引擎在特定生命周期节点触发回调外部代码可以在回调里执行自定义逻辑而不需要修改框架内部代码。Agent 工作流常见的钩子位置包括before_step步骤执行前可以用来做输入校验、权限检查、埋点after_step步骤执行后可以用来记录结果、更新上下文、发送通知on_error步骤出错时可以用来告警、记录堆栈、执行补偿逻辑before_llm_call / after_llm_call在调用模型前后介入比如注入上下文模板、校验模型返回的 JSON。钩子和中间件容易混淆。中间件通常在整个请求链路里按顺序处理覆盖范围更大钩子更强调“在某个节点上触发”粒度更细。实际项目里两者经常搭配使用中间件负责租户识别、日志链路钩子负责具体步骤的前置检查和后置处理。使用钩子最需要注意的是异常隔离。钩子里的日志采集代码不应该因为网络抖动就让整个工作流失败。推荐把钩子执行包在独立 try/except 里或者约定“钩子异常只告警不中断”除非是权限校验这类必须阻断的钩子。1.3 技能把重复工作沉淀成可复用能力技能Skill解决的是复用问题。某类工作经常重复执行比如“整理会议纪要”“把英文需求翻译成开发任务”如果每次都用一大段提示词重新描述既浪费 token 又不稳定。更好的做法是把这些重复流程固定成“技能”一个名称、一段触发说明、一组参数、一段执行逻辑。可以这样理解技能的结构描述description告诉模型什么时候该用这个技能参数parameters声明需要传入的字段、类型、默认值和约束执行逻辑handler真正干活的代码可以是函数、脚本或另一个工作流返回格式稳定的输出结构方便后续节点继续处理。为什么技能不能只是一个提示词模板因为提示词模板只是“说”技能是“说明 执行”。模型判断出应该调用某个技能后真正执行的是注册好的函数返回结果再交给模型继续处理。这样一来同样一段处理逻辑可以在多种场景复用而不需要反复编写相同提示词。热门搜索里经常出现“把重复工作流程保存为自定义技能”“固定指令模板”这类问题本质上就是在做技能化改造。起步时可以从一个最简单的技能开始把某个固定格式的文本处理封装成函数配上一段清晰的描述然后在工作流中注册验证模型能否在合适场景下正确调用。1.4 MCP 服务统一 Agent 和外部工具的接入方式MCP 是 Model Context Protocol 的缩写它解决的是工具接入标准化问题。没有统一协议之前每个工具都要为 Agent 定制一套接入方式有的走 REST有的走 SDK有的需要特殊鉴权。Agent 每接入一个新工具就要重新适配成本很高。MCP 的思路是定义一套通用协议让工具提供方实现一个 MCP Server负责描述自己有哪些工具、工具的参数是什么Agent 侧通过 MCP Client 发现工具、调用工具然后把工具结果返回给模型。类比来看它像是工具接入层的通用接口标准类似硬件里的 USB-C。从协议执行角度看核心是两个操作发现工具MCP Client 向 MCP Server 请求工具列表拿到每个工具的名称、描述和参数 schema调用工具MCP Client 按名称和参数调用某个工具Server 执行并返回结构化结果。注意MCP 的协议细节和官方 SDK 版本迭代比较快传输方式也可能不同。学习时不要只记某个版本的命令而要理解“发现工具 调用工具”这两个核心机制落地时再以官方文档为准。2. 在快报里看到项目后先评估这五个方面再决定是否深入2.1 判断项目是框架、示例集合还是可部署应用GitHub 快报里的项目分为几类用途完全不同。看到一个新仓库时第一步不是急着点 Star而是先判断它属于哪种定位。项目定位典型特征适合人群学习方式框架/引擎提供编程接口、插件点、配置规范想在自己系统里集成 Agent 能力的开发者跑通官方示例再看扩展点示例集合以 examples 或 demo 为主代码量小想理解某个概念的初学者逐个运行改参数看差别可部署应用带 Dockerfile、部署文档、完整 UI/API想直接使用的业务团队按部署文档启动再做配置适配定位判断会影响投入时间。框架类项目需要深入理解 API 设计示例集合适合快速建立直觉可部署应用则要重点关注数据、权限和运维成本。2.2 值得优先打开的文件拿到一个新项目后不要先浏览全部代码按顺序打开下面几类文件README看前两段能否说清楚项目解决什么问题、快速开始命令是否可复制examples 或 demo看是否有最小可运行示例这是判断项目是否好上手的直接依据hooks、plugins、extensions 相关目录看扩展点怎么设计决定能否接入自己的业务逻辑skills 或 tools 目录看技能/工具如何注册判断是否符合你的场景tests 目录有测试说明项目对稳定性有基本要求协议或 SDK 版本声明看依赖是否锁定避免踩版本坑。一个典型仓库结构可能长这样agent-project/ ├── README.md ├── examples/ │ └── simple_workflow.py ├── src/ │ ├── engine.py │ ├── hooks.py │ └── skills/ │ ├── __init__.py │ └── summary.py ├── mcp_server/ │ └── server.py └── tests/ └── test_engine.py看到这个结构对项目的扩展方式会有基本判断技能放在 skills 目录钩子逻辑在 hooks 里定义MCP 服务独立成模块。后续接入时只需要关注这几个入口。2.3 成熟度判断Star 数不等于工程质量Agent 方向的热门项目 Star 数涨得很快但 Star 只能说明关注度不能说明是否适合生产使用。判断成熟度至少要看四件事评估维度观察点活跃度最近提交时间、release 间隔版本稳定性是否进入 1.x是否有 breaking change 说明依赖约束是否锁定依赖版本是否给出兼容矩阵文档质量快速开始是否可运行配置项是否有解释测试覆盖核心引擎是否有单测示例是否可验证如果项目长期不维护、依赖没有锁定、示例跑不通即使 Star 很高也只建议作为参考不建议直接集成。3. 一个可运行的最小 Agent 工作流带钩子和技能3.1 先明确这个示例要演示什么为了把前面概念落到代码里这里实现一个最小可运行示例。目标不是实现完整 Agent而是演示三个机制技能如何注册和调用钩子如何在工作流节点上触发工作流如何按步骤顺序执行并保存结果。示例使用 Python 3.10 以上版本不依赖第三方大模型框架LLM 调用处用占位函数代替重点是流程本身。目录结构如下agent_demo/ ├── hooks.py ├── skills.py ├── engine.py └── main.py3.2 技能模块定义职责单一的可调用单元# skills.py from dataclasses import dataclass from typing import Any, Callable dataclass class Skill: name: str description: str params: dict handler: Callable[..., Any] SKILLS: dict[str, Skill] {} def register_skill(skill: Skill) - None: SKILLS[skill.name] skill def run_skill(name: str, **kwargs: Any) - Any: skill SKILLS.get(name) if skill is None: raise KeyError(fskill not found: {name}) return skill.handler(**kwargs) def _format_task(title: str, priority: str medium) - dict: return {title: title, priority: priority, status: created} def _append_note(task_id: str, note: str) - dict: return {task_id: task_id, note: note, status: updated} def init_skills() - None: register_skill(Skill( nameformat_task, description根据标题和优先级创建任务结构, params{title: {type: string, required: True}, priority: {type: string, default: medium}}, handler_format_task, )) register_skill(Skill( nameappend_note, description给指定任务追加备注, params{task_id: {type: string, required: True}, note: {type: string, required: True}}, handler_append_note, ))这里的关键是每个技能只做一件事并且参数定义明确。技能描述会被上层的模型推理逻辑用来决定是否调用所以描述里要写清楚“输入什么、输出什么”。这套结构后续可以换成任何 Agent 框架的技能注册方式思路一致。3.3 钩子模块隔离业务逻辑与横切逻辑# hooks.py from dataclasses import dataclass, field from typing import Callable def _log_before(context: dict) - None: step context[step] print(f[hook before_step] {step.get(name)} 准备执行) def _log_after(context: dict) - None: step context[step] print(f[hook after_step] {step.get(name)} 执行完成结果: {context.get(result)}) def _log_error(context: dict) - None: step context[step] print(f[hook on_error] {step.get(name)} 发生异常: {context.get(error)}) dataclass class HookRegistry: before_step: list[Callable[[dict], None]] field(default_factorylist) after_step: list[Callable[[dict], None]] field(default_factorylist) on_error: list[Callable[[dict], None]] field(default_factorylist) def register(self, event: str, fn: Callable[[dict], None]) - None: if event not in (before_step, after_step, on_error): raise ValueError(funknown hook event: {event}) getattr(self, event).append(fn) def trigger(self, event: str, context: dict) - None: for fn in getattr(self, event): try: fn(context) except Exception: # 钩子异常不能影响主流程记录后继续 print(f[hook failure] event{event})钩子模块的设计要点是异常隔离。这里用 try/except 把每个回调包起来即使某个埋点函数出了问题也不会让工作流节点中断。生产环境应该把这个 print 替换成结构化日志框架并记录 trace_id。3.4 工作流引擎按步骤顺序执行并维护上下文# engine.py from hooks import HookRegistry from skills import run_skill class AgentEngine: def __init__(self, hooks: HookRegistry): self.hooks hooks def run(self, steps: list[dict], context: dict) - dict: for step in steps: self.hooks.trigger(before_step, {step: step, context: context}) try: result self._execute(step, context) context[step[name]] result self.hooks.trigger(after_step, {step: step, result: result, context: context}) except Exception as exc: self.hooks.trigger(on_error, {step: step, error: exc, context: context}) raise return context def _execute(self, step: dict, context: dict) - dict: step_type step[type] if step_type skill: return run_skill(step[skill], **step.get(params, {})) if step_type llm: # 真实项目中这里替换为对模型服务的调用 prompt step[template].format(**context) return {output: fLLM 模拟输出: {prompt}} raise ValueError(funsupported step type: {step_type})工作流引擎只关心两件事顺序执行步骤、维护上下文。每个步骤的结果都会写回 context后续步骤可以用step[template].format(**context)引用前面的输出。这种设计与常见 Agent 框架里的节点上下文设计一致。3.5 入口与运行验证# main.py from hooks import HookRegistry, _log_before, _log_after, _log_error from engine import AgentEngine from skills import init_skills def main() - None: init_skills() hooks HookRegistry() hooks.register(before_step, _log_before) hooks.register(after_step, _log_after) hooks.register(on_error, _log_error) engine AgentEngine(hooks) steps [ {name: create_task, type: skill, skill: format_task, params: {title: 整理周报, priority: high}}, {name: add_note, type: skill, skill: append_note, params: {task_id: task-1, note: 待补充数据}}, {name: summarize, type: llm, template: 任务{create_task}备注{add_note}请生成一句跟进建议}, ] result engine.run(steps, {}) print(最终上下文, result) if __name__ __main__: main()运行cd agent_demo python main.py预期输出[hook before_step] create_task 准备执行 [hook after_step] create_task 执行完成结果: {title: 整理周报, priority: high, status: created} [hook before_step] add_note 准备执行 [hook after_step] add_note 执行完成结果: {task_id: task-1, note: 待补充数据, status: updated} [hook before_step] summarize 准备执行 [hook after_step] summarize 执行完成结果: {output: LLM 模拟输出: 任务{\title\: \整理周报\, ...}备注{...}请生成一句跟进建议} 最终上下文 {create_task: {...}, add_note: {...}, summarize: {...}}验证时关注三点步骤是否按声明顺序执行每个步骤结果是否写回 context钩子是否在正确节点触发。如果某个步骤抛异常on_error 钩子会打印错误主流程中断并抛出原始异常这是预期行为。4. MCP 服务搭建与调用流程从一个最小 Demo 说起4.1 先理解 MCP 的两步核心操作MCP 服务的调用流程本质上可以压缩成两步发现工具、调用工具。无论使用什么语言和 SDKclient 和 server 之间的消息大体是 JSON-RPC 风格。一个请求“列出所有工具”的消息大概长这样{ jsonrpc: 2.0, id: 1, method: tools/list, params: {} }服务端返回工具列表每个工具包含名称、描述和参数 schema{ jsonrpc: 2.0, id: 1, result: { tools: [ { name: get_current_time, description: 返回服务器当前时间, inputSchema: { type: object, properties: {} } } ] } }调用工具时发送tools/call{ jsonrpc: 2.0, id: 2, method: tools/call, params: { name: get_current_time, arguments: {} } }先理解这个流程再去读某个 SDK 的文档会容易很多。协议版本、传输方式和 SDK 命名都在变化但“列工具、调工具、返回结构化结果”这条主线是稳定的。4.2 一个最小 MCP Server Demo下面写一个不绑定具体 SDK 版本的 MCP Server 思路示例用于理解服务端需要提供什么能力。实际项目可以采用官方 Python SDK、TypeScript SDK或者自己按协议实现消息处理。# mcp_server_demo.py import datetime TOOLS { get_current_time: { description: 返回服务器当前时间, inputSchema: { type: object, properties: { timezone: {type: string, description: 时区可选} } }, }, add: { description: 计算两个整数之和, inputSchema: { type: object, properties: { a: {type: integer}, b: {type: integer} }, required: [a, b] }, }, } def _get_current_time(arguments: dict) - dict: return {ok: True, result: datetime.datetime.now().isoformat()} def _add(arguments: dict) - dict: a arguments.get(a, 0) b arguments.get(b, 0) return {ok: True, result: a b} HANDLERS { get_current_time: _get_current_time, add: _add, } def handle_request(method: str, params: dict) - dict: if method tools/list: return { tools: [ {name: name, description: meta[description], inputSchema: meta[inputSchema]} for name, meta in TOOLS.items() ] } if method tools/call: name params[name] arguments params.get(arguments, {}) handler HANDLERS.get(name) if handler is None: return {ok: False, error: funknown tool: {name}} return handler(arguments) return {ok: False, error: funsupported method: {method}}这个 demo 把 schema 和 handler 分开定义。tools 字典描述“能力清单”handlers 字典存放真正的执行函数。落地时要以这句话为验收基准schema 里声明什么参数handler 里就必须能接收什么参数两者不能脱节。4.3 客户端接入路线在本地验证 MCP 时开发路径通常是选择一个 MCP SDK确认协议版本和传输方式启动 MCP Server确认进程或端口正常用客户端发送 tools/list确认工具列表返回用客户端发送 tools/call传入最小参数确认返回结构把工具结果注入到 Agent 上下文中由模型决定下一步动作。在常见 Agent 平台或开发框架中例如 Dify、Coze、Trae、Cursor 等技能和 MCP 的接入形式各有差异有的提供可视化界面有的要求编写 manifest 文件有的需要调用 SDK。但底层逻辑一致先让平台知道“有什么工具、参数是什么”再让平台调用工具并把结果传给模型。4.4 验证方式不建议一上来就把 MCP 服务接入完整 Agent。先用最小客户端验证两条消息tools/list 和 tools/call。拿到稳定的工具列表和正确返回后再考虑接入工作流。这样排查问题时范围很小不会出现“不知道是 Agent 配置错还是 MCP 服务错”的情况。注意验证 MCP 服务时不要只看 tools/list 能返回还要用最小参数实际调用每个工具确认返回结构、错误分支和类型约束都符合预期。5. 实际项目中这几个地方最容易出问题5.1 钩子抛异常导致主流程中断现象只是加了一个日志钩子结果工作流经常在中间某一步中断日志里出现钩子报错。原因钩子回调直接执行了外部调用比如写日志、发通知外部系统抖动时异常向上抛出。检查方式看堆栈是否指向钩子函数临时禁用钩子后主流程是否恢复正常。处理方式钩子执行必须和业务执行隔离默认只记录异常不中断流程权限校验类必须阻断的钩子单独标记。参考 3.3 中的设计把每个回调包进独立的 try/except。预防建议为钩子定义统一接口限制回调签名禁止在钩子里做重操作或长时间等待。5.2 技能描述不准确Agent 从不调用现象技能注册成功但模型在处理本应该调用该技能的任务时直接输出文本答案不触发技能。原因技能描述写得过于宽泛模型不知道何时触发参数缺少默认值和约束模型生成参数时犹豫。检查方式把技能描述和参数说明打印出来站在模型角度判断是否清楚在测试里给一个明确触发指令看是否调用。处理方式描述里写清“当用户提出某类请求时使用”参数给默认值、类型和示例。例如“当用户需要创建任务时使用 format_task 技能”。预防建议每个技能只解决一个场景描述不超过两三句话参数数量控制在必要范围。5.3 MCP 工具参数契约不一致现象tools/call 返回参数校验错误或服务端收到 None。原因服务端 inputSchema 声明了必填参数但 handler 里没有做兼容处理客户端传入了 schema 之外的字段类型不匹配比如 schema 声明 integer客户端传了字符串。检查方式把实际发送的消息和 schema 放在一起对比在 handler 入口打印原始 arguments。处理方式以服务端 inputSchema 为唯一契约handler 内部再做强校验不要在客户端私自扩展字段。对每个工具至少写一个最小调用用例。预防建议工具返回统一用{ok: bool, result: ..., error: ...}结构方便上层判断成功失败。5.4 工作流缺少状态保存和幂等控制现象工作流重试后外部 API 被重复调用产生重复通知或重复数据写入。原因每个步骤都没有执行记录也没有幂等键失败重试时从头执行。检查方式查看重试后的调用记录确认是否重复请求了同一外部接口。处理方式给每个步骤分配 step_id把执行状态、输入、输出、耗时持久化对外部写操作传入幂等键同一请求重复提交时直接返回既有结果。预防建议在工作流引擎层设计状态存储抽象而不是在各个步骤里各自保存。6. 筛选项目和生产落地检查清单6.1 引入开源项目前的评估清单看到快报或热帖里的 Agent 项目先不要急着写代码。先用 15 分钟回答下面这些问题README 是否在开头说明项目解决什么问题是否有可复制的快速开始命令是否提供真实可运行的示例项目是框架、示例还是可部署应用扩展点钩子、技能、插件是否有文档MCP 支持是否有独立目录和版本说明依赖是否锁定版本是否给出兼容要求是否有测试最近一个月是否有提交许可证是否允许在目标场景使用如果核心问题多数是否建议只作为概念参考不要直接集成。6.2 从 Demo 走向生产的落地清单本地跑通之后生产环境需要补的东西比 demo 多得多配置外置协议地址、模型密钥、超时时间全部从配置文件或环境变量读取日志与链路给每次工作流执行分配 trace_id记录每个节点的输入输出摘要权限控制技能和工具按角色授权不能让任意用户调用任意工具超时与重试每个步骤设置超时时间写操作使用幂等键敏感信息保护密钥、个人数据不要写进提示词模板或技能描述结构校验对模型返回和工具返回做 schema 校验失败时走补偿逻辑监控告警记录成功率、耗时、token 消耗设置异常告警回滚方案技能或工作流变更时保留旧版本支持快速回退。6.3 新手的练习顺序跑通 3.3 到 3.5 的钩子和技能示例熟悉注册、触发、异常隔离单独搭一个 MCP Server用 tools/list 和 tools/call 两条消息验证把 MCP 工具包装成一个技能接入工作流完成最小闭环再对比 Dify、Coze、Trae、Cursor 或 LangGraph 等平台和框架理解它们的界面和 API 虽然在变底层都围绕这几个机制展开最后选一个真实业务场景把重复流程技能化并补上监控、重试和回滚。7. 扩展方向与下一步该关注什么7.1 从单 Agent 工作流到多 Agent 协作当前示例是单 Agent 顺序执行。真实系统里一个 Agent 往往不够有的负责规划有的负责执行有的负责质检。多 Agent 协作会在工作流之上增加通信机制例如消息队列、共享内存、事件总线。此时钩子变得更加重要因为每个 Agent 的启动、等待、失败都需要被观测。理解单 Agent 的钩子和技能机制后再学习多 Agent 协作会顺很多。7.2 技能和 MCP 的边界设计技能和 MCP 服务容易混在一起。推荐按这个边界划分技能是“业务动作的封装”由当前项目定义面向特定场景MCP 服务是“通用能力的接入”可以被多个项目共享。同一个功能可以先用技能实现验证价值后再抽象成 MCP 服务避免一开始就陷入协议细节。这样可以控制复杂度也让组件边界更清晰。7.3 长期维护需要补的工程能力Agent 项目上线后维护成本集中在三个方面模型输出不稳定、外部工具变更、依赖升级。对应的做法是给模型输出加 schema 校验和版本化给工具接口做兼容层给 SDK 和依赖做版本锁定和升级测试。这三点补齐后Agent 工作流才能真正进入稳定迭代而不是每改一次就重新排查一遍。实际项目中最值得投入精力的地方不是把工作流写得更花哨而是把每个技能的边界、每个钩子的异常处理、每个工具的契约定义清楚。这三件事做好Agent 系统的稳定性会明显提升。