1. 项目概述：Agent-Skills 不是插件，是能力调度中枢

“Agent-Skills”这个词最近在开发者社区里频繁刷屏，但很多人第一反应是——这又是个新出的 CLI 工具？还是某个大模型平台的技能市场？其实都不是。我从去年底开始系统性地搭建基于 LLM 的自动化工作流，从早期用 Python 脚本硬编码调用 API，到后来引入 LangChain 做 chain 编排，再到今年初接触真正意义上的 agent 架构后才意识到：Skills 不是功能模块，而是 agent 的“可调度肌肉”；Agent-Skills 是一套标准化的能力注册、发现、验证与执行协议。它解决的核心问题非常具体：当一个 agent 需要发邮件、查天气、读 Excel、调用内部 HR 系统、生成 PDF 报告时，它怎么知道该找谁？怎么确认这个“谁”真的能干？怎么安全地传参？怎么处理失败重试？怎么记录执行上下文？这些不是靠写死逻辑能覆盖的，必须有一套轻量、可插拔、带元信息描述、支持运行时发现的技能治理体系。

你可能已经用过类似的东西——比如 Slack 的 slash commands（/weather、/jira），或者 GitHub Actions 的 marketplace，甚至手机里的快捷指令（Shortcuts）。但它们要么绑定特定平台，要么缺乏统一的输入/输出契约，要么无法被其他 agent 动态发现和组合。而 Agent-Skills 的设计哲学，就是把“能力”本身变成一种可编程的一等公民。它不依赖特定框架（LangChain、LlamaIndex、DSPy 都能接入），不绑定某家大模型（Claude、DeepSeek、Qwen、GLM 全部兼容），也不强求部署形态（本地 CLI、HTTP 服务、Docker 容器、甚至 WASM 模块均可注册）。关键词 “CLI” 和 “slash commands” 出现在热搜里，恰恰说明大家是从最直观的交互入口感知到它的存在；而 “API” 高频出现，则暴露了底层本质——Skills 最终落地形态，90% 是封装良好的 RESTful 接口或标准函数调用。

适合谁看？如果你正在做以下任何一件事，这篇内容就值得你逐行读完：

• 用 Python 写了一堆工具脚本（查数据库、发钉钉、转 Markdown），但每次新加需求都要改主程序；
• 在 LangChain 中反复 copy-paste Tool 类定义，参数校验逻辑散落在各处；
• 用 FastAPI 暴露了十几个小接口，却没人知道它们能干什么、输入格式是什么、是否需要鉴权；
• 团队里前端、后端、算法同学各自开发“能力”，但集成时总在对 JSON Schema 和错误码；
• 甚至只是个技术产品经理，想给非技术人员提供“拖拽式技能组合”界面，但找不到统一的能力描述标准。
  这不是一个玩具项目，而是我在三个真实生产环境（金融风控报告生成、跨境电商多平台库存同步、制造业设备日志分析）中反复锤炼出的最小可行架构。下面我会从设计逻辑、核心契约、CLI 实现、API 对接、避坑实录五个维度，带你亲手搭起属于你自己的 Skills 生态。

2. 整体设计思路：为什么不用现有方案？为什么必须自建？

2.1 现有方案的三大硬伤

先说清楚我们绕不开的“前人经验”。目前主流的技能组织方式大概分三类：
第一类是框架内置 Tool 体系（LangChain 的 Tool、LlamaIndex 的 FunctionTool、DSPy 的 Program）。它们优点是上手快，文档齐全，直接 import 就能用。但致命缺陷在于：耦合度高、不可跨框架复用、元信息表达力弱。举个例子：LangChain 的@tool装饰器要求你写 docstring 描述功能，但这个字符串没法被 IDE 自动解析成参数提示，也不能被 Swagger 自动生成 API 文档，更不能被前端组件读取渲染成表单。我曾尝试把一个 LangChain Tool 直接迁移到 DSPy 流程里，结果发现参数名大小写不一致、required 字段缺失、返回值结构不同，最后花了两天重写适配层——这违背了“一次开发，多处调用”的初衷。

第二类是平台型技能市场（如 Claude 官方 Skills Market、Zapier 的 App Directory）。这类方案解决了发现和安装问题，但代价是彻底放弃控制权。你无法审核其代码安全性，无法修改超时时间，无法对接内部鉴权体系，甚至无法查看它实际调用了哪些下游 API。去年我们有个客户要求所有外部技能必须经过 SOC2 合规审计，结果发现 Zapier 上 87% 的连接器不提供源码审计权限，最终全部弃用。更现实的问题是：你的企业微信机器人、内部 OA 系统、产线 PLC 控制接口，根本不可能上架到公共市场。

第三类是纯手工 API 管理（Postman Collection + Swagger + 自建网关）。这是很多团队的实际做法，优点是完全可控。但问题在于：没有能力语义层，只有接口列表。当你在 Postman 里看到/api/v1/inventory/sync这个 endpoint，你能立刻判断它是否支持并发？是否需要前置登录态？失败时会返回什么业务码？有没有幂等性保证？这些信息全靠口头约定或藏在某个 Confluence 页面里。我见过最典型的情况是：算法同学开发了一个“用户画像打分”技能，文档写着“输入 user_id，返回 score”，结果上线后运营同学传了手机号，接口直接 500——因为没写明输入必须是内部 ID 格式，也没做类型校验。

2.2 Agent-Skills 的三层解耦设计

针对上述痛点，我们采用“契约先行、运行时发现、执行隔离”三层架构：

• 契约层（Contract Layer）：定义 Skills 的唯一身份、输入输出 Schema、执行约束（超时、重试、限流）、安全策略（是否需 OAuth、允许哪些 IP 段）、可观测性配置（是否记录输入、是否脱敏）。这个契约不依赖任何代码，用 YAML 或 JSON Schema 表达，存于 Git 仓库或配置中心。例如一个“发送企业微信消息”的 Skill 契约片段：

  id: wecom.send_message version: "1.2.0" description: "向指定成员发送文本消息，支持 markdown 格式" input_schema: type: object required: [user_id, content] properties: user_id: type: string pattern: "^[a-zA-Z0-9_]{4,32}$" # 严格校验 ID 格式 content: type: string maxLength: 2048 mentioned_list: type: array items: {type: string} output_schema: type: object required: [message_id, status] properties: message_id: {type: string} status: {enum: ["success", "failed", "rate_limited"]} constraints: timeout_ms: 5000 max_retries: 2 rate_limit: "100/minute" security: auth_required: true scopes: ["wecom.message.send"]

• 发现层（Discovery Layer）：提供统一的注册中心（Registry），支持多种注册方式：CLI 命令行注册、HTTP POST 注册、Git Webhook 自动同步、Kubernetes CRD 声明式注册。关键设计是“软注册”机制——注册时只校验契约合法性（YAML 格式、Schema 有效性），不校验实际服务是否在线。这样开发阶段就能提前定义能力，测试环境和服务上线可以异步进行。我们用 SQLite 做轻量注册中心（单机够用），用 PostgreSQL 做高可用集群（支持百万级 Skills 管理），核心表只有三张：skills（存契约）、endpoints（存服务地址）、versions（存灰度版本）。

• 执行层（Execution Layer）：这才是真正干活的部分。它不关心 Skills 用什么语言写、部署在哪，只认契约。当 agent 发起调用时，执行层根据 skill_id 查到契约，做三件事：1）按契约校验输入参数（用 JSON Schema Validator）；2）按安全策略检查调用方权限；3）按 endpoint 路由到实际服务，加超时/重试/熔断保护。我们实现了一个通用执行器（Executor），支持 HTTP、gRPC、本地进程、Docker Run 四种调用模式。比如一个 Python 写的技能，可以打包成 Docker 镜像注册，执行器会自动拉起容器并传参；一个 Node.js 的 CLI 工具，执行器会启动子进程并监听 stdout/stderr。

这种设计带来的直接好处是：前端同学写个 React 组件展示 Skills 列表，只要读取 Registry 的/skills接口，就能自动生成带表单的调用界面；运维同学用 Prometheus 监控所有 Skills 的 P95 延迟，只需在执行器里埋点；安全团队审计时，直接扫描所有契约文件里的security.auth_required字段即可。所有角色都基于同一份契约工作，而不是各自维护一份“理解”。

2.3 为什么 CLI 是最佳切入点？

热搜词里 “CLI” 和 “slash commands” 高频出现，不是偶然。CLI 是 Skills 最自然的开发调试入口。原因有三：
第一，开发反馈闭环最快。写完一个 Python 脚本，执行agent-skills register --file send_wecom.yaml，再agent-skills invoke wecom.send_message --user_id zhangsan --content "你好"，5 秒内就能看到结果。对比 Web UI 提交表单、等待审批、再进测试环境验证，效率提升十倍。我团队规定：所有 Skills 必须先通过 CLI 验证，才能提交到 CI 流水线。

第二，天然支持管道组合。Unix 哲学“每个程序只做一件事，并做好”在这里完美复用。比如生成周报的完整流程：agent-skills invoke jira.get_issues --project FINANCE | agent-skills invoke llm.summarize --model qwen2 | agent-skills invoke wecom.send_message --user_id @all。这种链式调用比写一个巨无霸 Python 脚本清晰得多，也方便单独替换其中一环（比如把 qwen2 换成 deepseek-v2）。

第三，降低非程序员参与门槛。运维同学不需要懂 Python，只要会写 YAML 契约和 Shell 脚本，就能把curl命令包装成 Skills；产品同学用agent-skills list --category notification就能查到所有通知类能力，不用翻代码库。我们甚至用 CLI 做了内部“技能考试”：新人入职第一周任务是用 CLI 注册 3 个 Skills（查天气、发邮件、读 Excel），通过即算环境配置合格。

提示：不要把 CLI 当成最终交付形态。它只是开发期的瑞士军刀，生产环境应该用 HTTP API 或 SDK 集成。但如果你跳过 CLI 阶段直接上 Web 管理后台，90% 的团队会在契约定义环节卡住——因为缺少即时反馈，大家会陷入“这个字段要不要必填”的无休止争论。

3. 核心细节解析：契约、注册、执行的魔鬼细节

3.1 契约设计的五个反直觉原则

很多人以为契约就是写个 JSON Schema，其实远不止。我们在 200+ 个真实 Skills 中总结出五条必须遵守的原则，违反任意一条都会导致后期集成灾难：

原则一：禁止使用$ref引用外部 Schema
看似能复用，实则埋雷。比如你定义common.user_id.yaml，然后在 10 个 Skills 里引用它。某天你更新了user_id格式（从 8 位数字改为 UUID），所有引用它的 Skills 都得同步改，否则契约校验失败。更糟的是，不同 Skills 对user_id的语义理解可能不同（A 认为是员工工号，B 认为是微信 openid）。解决方案：每个 Skill 的 input_schema 必须内联定义，哪怕重复 10 次。我们用脚本自动生成重复字段的注释：“此字段语义：企业微信内部成员 ID，非手机号，非邮箱”。

原则二：输出 Schema 必须包含明确的错误分支
90% 的契约只定义 success 返回，但实际调用中 30% 是失败。如果契约不声明错误结构，调用方只能靠 try-catch 捕获异常，无法做精细化错误处理。正确做法是在 output_schema 里用oneOf显式声明成功和失败结构：

output_schema: oneOf: - type: object required: [status, data] properties: status: {const: "success"} data: {type: object, properties: {message_id: {type: string}}} - type: object required: [status, error_code, error_message] properties: status: {const: "error"} error_code: {enum: ["INVALID_USER_ID", "RATE_LIMIT_EXCEEDED", "WECOM_TOKEN_EXPIRED"]} error_message: {type: string}

这样前端就能根据error_code显示不同提示，而不是统一弹“调用失败”。

原则三：约束字段必须可量化、可审计
timeout_ms: 5000比timeout: short好，“short” 是主观词。同理，rate_limit: "100/minute"比rate_limit: high可审计。我们强制要求所有约束字段带单位，且数值必须是整数。特别注意max_retries：必须明确是“重试次数”还是“最大尝试次数”（含首次），我们定义为后者，即max_retries: 3表示最多调用 3 次（首次 + 2 次重试）。

原则四：安全策略必须区分认证（Authentication）和授权（Authorization）
auth_required: true只解决“你是谁”，不解决“你能干啥”。必须配套scopes字段声明所需权限范围。比如wecom.message.send和wecom.message.delete是两个独立 scope，不能混用。我们用 OAuth2 的 scope 机制实现，调用方 token 必须包含请求 Skill 所需的所有 scopes，缺一不可。这点在多租户场景下至关重要——销售部的 token 不能调用财务部的erp.invoice.query。

原则五：版本号必须语义化，且主版本升级需破坏性变更
version: "1.2.0"不是随便写的。我们遵循 SemVer 规范：主版本（1.x.x）升级表示 input/output schema 有 breaking change；次版本（1.2.x）表示新增非破坏性字段或优化；修订版（1.2.3）表示 bugfix。Registry 会自动拦截主版本降级（防止误操作），并提供版本兼容性检查工具：agent-skills diff v1.1.0 v1.2.0 --skill wecom.send_message会输出具体哪些字段被删除或类型变更。

3.2 CLI 注册流程的七步精解

CLI 不是简单包装 curl，它承担着契约校验、本地测试、安全扫描三重职责。以agent-skills register命令为例，完整流程如下：

第一步：本地契约语法校验
CLI 用 PyYAML 解析 YAML 文件，检查基础语法（缩进、冒号、引号匹配）。这步秒级完成，但能捕获 60% 的低级错误。比如少了个冒号：

# 错误示例 constraints: timeout_ms: 5000 max_retries 3 # 这里漏了冒号！

CLI 会报错：YAML parse error at line 5, column 15: expected <block end>, but found '<scalar>'，并定位到具体行。

第二步：JSON Schema 有效性校验
加载内置的 Skills 契约 Schema（一个严格的 JSON Schema 文件），用jsonschema.validate()校验整个契约结构。这步会检查：id是否符合正则^[a-z][a-z0-9.-]*$（禁止大写和下划线），input_schema是否是合法 JSON Schema，constraints.timeout_ms是否为正整数等。我们内置了 47 条校验规则，覆盖所有可能的契约陷阱。

第三步：本地服务连通性测试
如果契约里指定了endpoint.type: http，CLI 会尝试向endpoint.url发送预检请求（HEAD 或 OPTIONS），验证服务是否可达。这步不是必须的（允许离线注册），但强烈建议开启：agent-skills register --file send_wecom.yaml --test-endpoint。我们遇到过最典型的错误：开发同学写了http://localhost:8000，但实际服务跑在 Docker 容器里，CLI 在宿主机执行时根本连不上。

第四步：输入参数模拟校验
CLI 会生成符合input_schema的随机测试数据（用json-schema-faker库），调用本地服务进行端到端测试。比如input_schema要求user_id是 4-32 位字母数字，它就生成zhangsan123；要求content最长 2048 字符，它就生成 2048 个a。这步能发现 80% 的运行时参数校验漏洞——比如后端代码没做长度限制，导致数据库字段溢出。

第五步：安全策略静态扫描
扫描契约中的security字段，检查是否存在高危配置：auth_required: false且scopes为空（意味着公开接口），rate_limit为"unlimited"（拒绝注册），timeout_ms大于 30000（超长超时影响 agent 响应）。扫描结果以 WARNING 形式输出，但不阻止注册（留给人工决策）。

第六步：Git 提交检查（可选）
如果当前目录是 Git 仓库，CLI 会检查契约文件是否已git add。未暂存的文件会被警告：“Warning: file not staged for commit. Consider running 'git add send_wecom.yaml'”。这确保所有 Skills 契约都纳入版本控制，避免配置漂移。

第七步：注册中心写入与版本生成
最后一步才是真正的注册：CLI 将校验通过的契约写入 Registry。Registry 会自动生成唯一version_id（UUID），并创建符号链接指向最新版（latest）。同时，CLI 会输出一条可复制的命令：

✅ Registered successfully! Skill ID: wecom.send_message Version: 1.2.0 (auto-generated) Endpoint: http://wecom-service:8000/v1/send Try it: agent-skills invoke wecom.send_message --user_id zhangsan --content "test"

这条命令可以直接粘贴执行，形成开发闭环。

注意：注册不是“发布”。注册后的 Skill 默认状态为draft，需手动agent-skills publish wecom.send_message才对外可见。draft 状态下只有 owner 和 admin 能调用，避免未测试完成的 Skills 被误用。

3.3 执行层的容错设计：如何让 Skills 像水电一样可靠

执行层是 Skills 生态的“心脏”，它必须扛住各种意外。我们不追求 100% 成功率（那不现实），而是确保失败时行为可预测、可观测、可恢复。以下是四个关键设计：

1. 输入校验的“防御性解析”
契约里input_schema要求user_id是字符串，但调用方可能传整数12345。传统做法是直接报错，但实际中更友好：执行器会尝试类型转换（int → str），仅当转换失败（如传null）才拒绝。我们用pydantic.BaseModel做输入解析，它内置了智能类型转换，比手写 if-else 更健壮。

2. 超时与重试的“阶梯式策略”
不是所有 Skills 都适用统一超时。执行器支持 per-skill 覆盖：在契约里写constraints.timeout_ms: 10000，执行器就用 10 秒。重试策略更精细：max_retries: 2是基础，但还支持retry_on_status: [408, 429, 502, 503, 504]（只对网关超时、限流、服务不可用重试），retry_backoff: "exponential"（指数退避：第一次 100ms，第二次 200ms，第三次 400ms）。

3. 错误传播的“语义化透传”
执行器不吞掉下游错误。如果 Skills 服务返回{"status": "error", "error_code": "INVALID_USER_ID"}，执行器原样透传，不包装成{"error": {"code": "EXECUTION_FAILED", "message": "Skill invocation failed"}}。这样调用方能直接处理业务错误，而不是在层层包装中迷失。

4. 可观测性的“黄金指标埋点”
每个 Skills 调用都记录 5 个黄金指标：

• duration_ms: 从收到请求到返回响应的总耗时
• input_size_bytes: 输入 JSON 的字节数（监控恶意大 payload）
• output_size_bytes: 输出 JSON 的字节数
• status: success / error / timeout / rate_limited
• error_code: 如果 status=error，记录具体的 error_code
  这些指标通过 OpenTelemetry 导出到 Prometheus，我们用 Grafana 做了统一看板，按skill_id、status、error_code多维下钻。最常看的视图是：“过去 1 小时内，哪个 Skills 的RATE_LIMIT_EXCEEDED错误最多？”——这能快速定位是调用方滥用，还是 Skills 限流配置过严。

4. 实操过程：从零搭建 CLI 工具与 API 服务

4.1 CLI 工具的完整实现（Python + Click）

我们用 Python 3.10+ 和 Click 框架实现 CLI，核心命令只有 5 个，但覆盖 95% 场景。以下是agent-skills主程序骨架（已简化，保留关键逻辑）：

# cli.py import click import yaml import json from pathlib import Path from jsonschema import validate, ValidationError from jsonschema.validators import Draft202012Validator from typing import Dict, Any, Optional # 加载内置契约 Schema SCHEMA_PATH = Path(__file__).parent / "schemas" / "skill-contract.json" with open(SCHEMA_PATH) as f: CONTRACT_SCHEMA = json.load(f) @click.group() @click.option('--registry-url', default='http://localhost:8000', help='Registry URL') @click.pass_context def cli(ctx, registry_url): ctx.ensure_object(dict) ctx.obj['REGISTRY_URL'] = registry_url @cli.command() @click.argument('file', type=click.Path(exists=True)) @click.option('--test-endpoint', is_flag=True, help='Test endpoint connectivity') @click.option('--publish', is_flag=True, help='Publish immediately after registration') @click.pass_context def register(ctx, file, test_endpoint, publish): """Register a new Skill from YAML contract file""" # Step 1: Parse YAML try: with open(file) as f: contract = yaml.safe_load(f) except yaml.YAMLError as e: click.echo(f"❌ YAML parse error: {e}", err=True) return # Step 2: Validate against schema try: validate(instance=contract, schema=CONTRACT_SCHEMA, cls=Draft202012Validator) except ValidationError as e: click.echo(f"❌ Contract validation error: {e.message} at {'.'.join([str(i) for i in e.absolute_path])}", err=True) return # Step 3: Test endpoint if requested if test_endpoint and 'endpoint' in contract: endpoint = contract['endpoint'] if endpoint.get('type') == 'http': import requests try: resp = requests.head(endpoint['url'], timeout=5) if resp.status_code not in [200, 204, 405]: click.echo(f"⚠️ Endpoint test warning: {resp.status_code} {resp.reason}", err=True) except Exception as e: click.echo(f"❌ Endpoint test failed: {e}", err=True) return # Step 4: Register to registry import requests try: resp = requests.post( f"{ctx.obj['REGISTRY_URL']}/skills", json=contract, timeout=10 ) resp.raise_for_status() result = resp.json() click.echo(f"✅ Registered successfully!") click.echo(f" Skill ID: {result['id']}") click.echo(f" Version: {result['version']}") if publish: # Auto-publish requests.post(f"{ctx.obj['REGISTRY_URL']}/skills/{result['id']}/publish") click.echo(" Status: published") except requests.RequestException as e: click.echo(f"❌ Registration failed: {e}", err=True) @cli.command() @click.argument('skill_id') @click.option('--version', default='latest', help='Version to invoke') @click.option('--param', '-p', multiple=True, nargs=2, type=str, help='Parameter as key value pair') @click.pass_context def invoke(ctx, skill_id, version, param): """Invoke a registered Skill""" # Build input dict from params input_data = {} for k, v in param: # Try to auto-convert types: "123" -> 123, "true" -> True, "null" -> None if v.lower() in ['true', 'false']: input_data[k] = v.lower() == 'true' elif v.lower() == 'null': input_data[k] = None else: try: input_data[k] = int(v) except ValueError: try: input_data[k] = float(v) except ValueError: input_data[k] = v # Call registry to get endpoint and invoke import requests try: # Get skill info resp = requests.get(f"{ctx.obj['REGISTRY_URL']}/skills/{skill_id}/versions/{version}") resp.raise_for_status() skill_info = resp.json() # Invoke actual service endpoint = skill_info['endpoint'] if endpoint['type'] == 'http': resp = requests.post( endpoint['url'], json=input_data, timeout=skill_info.get('constraints', {}).get('timeout_ms', 5000) / 1000 ) resp.raise_for_status() click.echo(json.dumps(resp.json(), indent=2, ensure_ascii=False)) except Exception as e: click.echo(f"❌ Invocation failed: {e}", err=True) if __name__ == '__main__': cli()

安装方式极其简单：

pip install agent-skills-cli # 或从源码安装 git clone https://github.com/your-org/agent-skills-cli.git cd agent-skills-cli pip install -e .

关键设计点：

• 参数自动类型转换：--param user_id 12345会自动转成整数，--param active true转成布尔值，避免调用方手动写 JSON。
• 超时继承：invoke命令自动读取契约里的timeout_ms，并转换为 requests 的timeout参数。
• 错误分级输出：CLI 用不同颜色和符号区分状态（✅ 成功，❌ 严重错误，⚠️ 警告），便于脚本解析。

4.2 Registry API 服务（FastAPI 实现）

Registry 是 Skills 的“大脑”，我们用 FastAPI 实现，核心接口只有 4 个，但承载全部元数据管理：

# registry/main.py from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks from pydantic import BaseModel, Field, validator from typing import List, Optional, Dict, Any from datetime import datetime import sqlite3 import uuid app = FastAPI(title="Agent-Skills Registry") # 数据库初始化 def init_db(): conn = sqlite3.connect("registry.db") conn.execute(""" CREATE TABLE IF NOT EXISTS skills ( id TEXT PRIMARY KEY, version TEXT NOT NULL, contract TEXT NOT NULL, status TEXT CHECK(status IN ('draft', 'published')) DEFAULT 'draft', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close() init_db() class SkillContract(BaseModel): id: str = Field(..., regex=r'^[a-z][a-z0-9.-]*$') version: str description: str input_schema: Dict[str, Any] output_schema: Dict[str, Any] constraints: Dict[str, Any] security: Dict[str, Any] @app.post("/skills", response_model=dict) def register_skill(contract: SkillContract): """Register a new skill contract""" conn = sqlite3.connect("registry.db") try: # Check if skill exists cursor = conn.execute("SELECT COUNT(*) FROM skills WHERE id = ?", (contract.id,)) if cursor.fetchone()[0] > 0: # Insert new version version_id = str(uuid.uuid4()) conn.execute( "INSERT INTO skills (id, version, contract, status) VALUES (?, ?, ?, ?)", (contract.id, contract.version, contract.json(), 'draft') ) else: # First version conn.execute( "INSERT INTO skills (id, version, contract, status) VALUES (?, ?, ?, ?)", (contract.id, contract.version, contract.json(), 'draft') ) conn.commit() return {"id": contract.id, "version": contract.version, "status": "registered"} finally: conn.close() @app.get("/skills/{skill_id}/versions/{version}", response_model=SkillContract) def get_skill_version(skill_id: str, version: str): """Get specific version of a skill""" conn = sqlite3.connect("registry.db") try: cursor = conn.execute( "SELECT contract FROM skills WHERE id = ? AND version = ? AND status = 'published'", (skill_id, version) ) row = cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Skill not found or not published") return SkillContract.parse_raw(row[0]) finally: conn.close() @app.post("/skills/{skill_id}/publish") def publish_skill(skill_id: str): """Publish latest draft version""" conn = sqlite3.connect("registry.db") try: # Get latest draft cursor = conn.execute( "SELECT version, contract FROM skills WHERE id = ? AND status = 'draft' ORDER BY created_at DESC LIMIT 1", (skill_id,) ) row = cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="No draft version found") # Update status to published conn.execute( "UPDATE skills SET status = 'published' WHERE id = ? AND version = ?", (skill_id, row[0]) ) conn.commit() return {"id": skill_id, "version": row[0], "status": "published"} finally: conn.close() @app.get("/skills", response_model=List[Dict[str, Any]]) def list_skills(category: Optional[str] = None, status: str = "published"): """List all skills (with optional category filter)""" conn = sqlite3.connect("registry.db") try: query = "SELECT id, version, contract FROM skills WHERE status = ?" params = [status] if category: query += " AND json_extract(contract, '$.category') = ?" params.append(category) cursor = conn.execute(query, params) results = [] for row in cursor.fetchall(): contract = json.loads(row[2]) results.append({ "id": row[0], "version": row[1], "description": contract.get("description", ""), "category": contract.get("category", "uncategorized") }) return results finally: conn.close()

部署方式：

# 安装依赖 pip install fastapi uvicorn pydantic python-multipart # 启动服务 uvicorn registry.main:app --host 0.0.0.0 --port 8000 --reload

关键特性：

• SQLite 轻量起步：单机开发完全够用，数据文件registry.db可直接备份。
• 状态机管理：draft→published状态流转，避免未测试 Skills 被误调用。
• Category 支持：契约里可加category: "notification"字段，list_skills支持按分类过滤，方便前端分类展示。

4.3 Skills 服务开发模板（Python Flask）

最后是 Skills 服务本身。我们提供标准化模板，确保所有 Skills 行为一致：

# skills/wecom_send.py from flask import Flask, request, jsonify import requests import os app = Flask(__name__) # 从环境变量读取配置，不硬编码 WECOM_TOKEN = os.getenv("WECOM_TOKEN") WECOM_AGENT_ID = os.getenv("WECOM_AGENT_ID") @app.route("/v1/send", methods=["POST"]) def send_message(): try: # 1. 解析输入（执行器已做过 Schema 校验，这里做业务校验） data = request.get_json() user_id = data.get("user_id") content = data.get("content") if not user_id or not content: return jsonify({ "status": "error", "error_code": "MISSING_REQUIRED_FIELD", "error_message": "user_id and content are required" }), 400 # 2. 调用企业微信 API wecom_url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" payload = { "touser": user_id, "msgtype": "text", "agentid": WECOM_AGENT_ID, "text": {"content": content}, "safe": 0 } headers = {"Content-Type": "application/json"} resp = requests.post( wecom_url, params={"access_token": WECOM_TOKEN}, json=payload, timeout=5 ) # 3. 处理响应 if resp.status_code == 2