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 Validator2按安全策略检查调用方权限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和授权Authorizationauth_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 Schemaconstraints.timeout_ms是否为正整数等。我们内置了 47 条校验规则覆盖所有可能的契约陷阱。第三步本地服务连通性测试如果契约里指定了endpoint.type: httpCLI 会尝试向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_idUUID并创建符号链接指向最新版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 的字节数监控恶意大 payloadoutput_size_bytes: 输出 JSON 的字节数status: success / error / timeout / rate_limitederror_code: 如果 statuserror记录具体的 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, defaulthttp://localhost:8000, helpRegistry URL) click.pass_context def cli(ctx, registry_url): ctx.ensure_object(dict) ctx.obj[REGISTRY_URL] registry_url cli.command() click.argument(file, typeclick.Path(existsTrue)) click.option(--test-endpoint, is_flagTrue, helpTest endpoint connectivity) click.option(--publish, is_flagTrue, helpPublish 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}, errTrue) return # Step 2: Validate against schema try: validate(instancecontract, schemaCONTRACT_SCHEMA, clsDraft202012Validator) except ValidationError as e: click.echo(f❌ Contract validation error: {e.message} at {..join([str(i) for i in e.absolute_path])}, errTrue) 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], timeout5) if resp.status_code not in [200, 204, 405]: click.echo(f⚠️ Endpoint test warning: {resp.status_code} {resp.reason}, errTrue) except Exception as e: click.echo(f❌ Endpoint test failed: {e}, errTrue) return # Step 4: Register to registry import requests try: resp requests.post( f{ctx.obj[REGISTRY_URL]}/skills, jsoncontract, timeout10 ) 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}, errTrue) cli.command() click.argument(skill_id) click.option(--version, defaultlatest, helpVersion to invoke) click.option(--param, -p, multipleTrue, nargs2, typestr, helpParameter 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], jsoninput_data, timeoutskill_info.get(constraints, {}).get(timeout_ms, 5000) / 1000 ) resp.raise_for_status() click.echo(json.dumps(resp.json(), indent2, ensure_asciiFalse)) except Exception as e: click.echo(f❌ Invocation failed: {e}, errTrue) 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(titleAgent-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(..., regexr^[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_modeldict) 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_modelSkillContract) 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_code404, detailSkill 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_code404, detailNo 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_modelList[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}, jsonpayload, timeout5 ) # 3. 处理响应 if resp.status_code 2