臭鼠标1. 项目概述
在AI技术快速发展的今天，智能体(Agent)的能力构建方式正在经历一场革命性的变革。传统的单体式AI系统开发模式已经无法满足复杂场景的需求，而基于"技能化"的设计思路正在成为行业主流。本文将详细介绍如何从零开始构建一套支持MCP协议的Agent Skills体系，实现智能体能力的模块化、标准化和可复用。
这个项目的核心价值在于：
- 将AI能力拆解为原子化的功能单元(Skills)
- 通过标准化协议(MCP)实现技能的发现和调用
- 构建"技能服务端+Agent客户端"的松耦合架构
- 提供完整的开源实现，可直接用于生产环境
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心设计理念
2.1 Agent Skills的四大设计原则
2.1.1 原子化设计
每个技能只负责单一功能，如查询天气、文本摘要或文件读取。这种设计带来三个显著优势：
- 复用性：技能可以在不同场景中被多次调用
- 可维护性：单个技能的修改不会影响其他功能
- 可测试性：每个技能可以独立测试和验证
2.1.2 标准化输入输出
所有技能都遵循统一的接口规范：
- 输入：明确的参数类型和结构
- 输出：固定格式的字典，包含执行状态、业务数据和提示信息
这种标准化设计极大降低了调用端的适配成本。
2.1.3 解耦设计
技能实现与调用逻辑完全分离：
- 技能层：只关注"如何实现功能"
- 调用层：只关注"何时调用哪个技能"
这种解耦使得技能开发和业务逻辑开发可以并行进行。
2.1.4 可注册/可发现/可管控
技能支持中心化注册和动态发现，便于：
- 统一管理所有可用技能
- 实时获取技能元数据
- 实施细粒度的权限控制
2.2 MCP协议的核心价值
MCP(Model Control Protocol)是专为AI Agent设计的技能调用协议，解决了三个关键问题：
2.2.1 技能服务化
将零散的技能封装为标准化服务，支持：
- 跨进程调用
- 跨语言集成
- 分布式部署
2.2.2 调用标准化
定义了统一的：
- 技能元数据格式
- 调用请求结构
- 响应数据结构
2.2.3 生态兼容性
使不同厂商的Agent能够：
- 通过统一协议调用技能
- 无缝集成第三方技能
- 打破技术生态壁垒
3. 技术架构实现
3.1 整体架构设计
项目采用经典的"技能服务端+Agent客户端"架构：
code复制┌─────────────────────┐ ┌─────────────────────┐
│ MCP技能服务端 │ │ MCP Agent客户端 │
│ （FastAPI实现） │ │ （DeepSeek大模型） │
│ 1. 技能注册 │ │ 1. LLM决策选技能 │
│ 2. 技能发现 │◄────►│ 2. MCP协议调技能 │
│ 3. 标准化技能调用 │ │ 3. 返回执行结果 │
└─────────────────────┘ └─────────────────────┘
│
▼
┌─────────────────────┐
│ 原子化技能层 │
│ （查天气/摘要/读文件）│
└─────────────────────┘
3.2 目录结构说明
项目采用模块化设计，主要目录结构如下：
code复制learn-agent-skills/
├── skills/ # 原子化技能实现
│ ├── base_skills.py # 核心技能实现
│ └── skill_registry.py # 技能注册中心
├── mcp_server/ # MCP服务端
│ ├── mcp_protocol.py # 协议定义
│ └── main_server.py # HTTP服务实现
├── mcp_agent/ # Agent客户端
│ ├── mcp_protocol.py # 协议定义
│ └── main_agent.py # 大模型决策逻辑
├── .env.template # 环境变量模板
├── pyproject.toml # 依赖配置
└── uv.lock # 依赖锁定文件
4. 核心代码解析
4.1 原子化技能实现
以天气查询技能为例，展示标准化技能的实现方式：
python复制def get_weather(city: str) -> Dict[str, str]:
"""原子化天气查询技能"""
try:
# 从环境变量获取API密钥
appid = os.getenv("WEATHER_APPID")
appsecret = os.getenv("WEATHER_APPSECRET")
# 构建API请求URL
api_url = f"http://v0.yiketianqi.com/api?unescape=1&version=v91&appid={appid}&appsecret={appsecret}&city={city}"
# 发送请求并处理响应
response = requests.get(api_url, timeout=10)
response.raise_for_status()
data = response.json()
# 标准化返回结果
return {
"city": city,
"temperature": data["data"][0]["tem"],
"weather": data["data"][0]["wea"],
"message": "查询成功"
}
except Exception as e:
# 错误处理也遵循标准化格式
return {"error": f"天气查询异常：{str(e)}"}
4.2 MCP协议定义
协议定义了三个核心数据结构：
python复制# 技能元数据
class MCPSkillMetadata(TypedDict):
name: str # 技能唯一标识
description: str # 功能描述
parameters: Dict[str, Any] # 参数定义
return_type: str # 返回类型
# 调用请求
@dataclass
class MCPInvokeRequest:
skill_name: str # 要调用的技能名
parameters: Dict[str, Any] # 调用参数
request_id: str = "" # 请求ID（用于追踪）
# 调用响应
@dataclass
class MCPInvokeResponse:
success: bool # 调用是否成功
data: Dict[str, Any] # 返回数据
error: str = "" # 错误信息
4.3 MCP服务端实现
基于FastAPI实现的两个核心接口：
python复制@app.get("/mcp/skills")
async def list_mcp_skills():
"""技能发现接口"""
return {"skills": list(MCP_SKILL_METADATA.values())}
@app.post("/mcp/invoke")
async def invoke_mcp_skill(request: MCPInvokeRequest = Body(...)):
"""技能调用接口"""
try:
# 获取并执行技能函数
skill_func = skill_registry.get_skill(request.skill_name)
result = skill_func(**request.parameters)
# 返回标准化响应
return MCPInvokeResponse(
success=True,
data=result,
request_id=request.request_id
).__dict__
except Exception as e:
return MCPInvokeResponse(
success=False,
error=str(e),
request_id=request.request_id
).__dict__
4.4 Agent客户端实现
客户端结合大模型决策和MCP调用：
python复制class MCPAgent:
def get_llm_decision(self, user_query: str) -> dict:
"""让大模型决定调用哪个技能"""
# 构建提示词，包含所有可用技能信息
prompt = f"""根据用户问题选择合适的技能：
{skills_info}
仅返回JSON格式的决策结果"""
# 调用大模型API获取决策
response = requests.post(DEEPSEEK_API_URL, json={
"messages": [{"role": "user", "content": prompt}]
})
return response.json()
def run(self, user_query: str):
"""执行完整流程"""
# 1. 大模型决策
decision = self.get_llm_decision(user_query)
# 2. MCP调用技能
result = self.mcp_client.invoke_skill(
decision["skill_name"],
decision["parameters"]
)
# 3. 返回结果
return {
"success": result.success,
"data": result.data,
"error": result.error
}
5. 部署与运行指南
5.1 环境准备
- 克隆项目仓库：
bash复制git clone https://gitee.com/impl/learn-agent-skills.git
cd learn-agent-skills
- 安装依赖：
bash复制uv sync
- 配置环境变量：
bash复制cp .env.template .env
# 编辑.env文件配置实际API密钥
5.2 启动服务端
bash复制uv run mcp_server/main_server.py
服务启动后，可以通过以下接口访问：
GET /mcp/skills- 获取所有可用技能POST /mcp/invoke- 调用指定技能
5.3 启动客户端
bash复制uv run mcp_agent/main_agent.py
客户端启动后会执行预设的测试用例，展示完整的工作流程。
6. 常见问题解决方案
6.1 Python模块导入问题
当出现ModuleNotFoundError时，需要在文件开头添加：
python复制import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
6.2 第三方API适配问题
建议采用防御性编程处理API响应：
python复制# 不推荐的写法（依赖固定字段）
if data["errcode"] == 0:
# 推荐的写法（通用字段检查）
if "data" in data and isinstance(data["data"], list) and len(data["data"]) > 0:
6.3 大模型决策优化
可以通过以下方式提升决策质量：
- 在提示词中提供更详细的技能描述
- 添加示例对话引导模型输出格式
- 设置适当的temperature参数控制输出稳定性
7. 扩展与优化建议
7.1 技能热加载机制
实现技能的动态加载和卸载，无需重启服务：
python复制@app.post("/mcp/register")
async def register_skill(skill_def: dict):
"""动态注册新技能"""
skill_registry.register(skill_def["name"], skill_def["func"])
return {"success": True}
7.2 权限控制系统
添加基于角色的访问控制：
python复制def check_permission(skill_name: str, user: User):
"""检查用户是否有权限调用该技能"""
return skill_name in user.allowed_skills
7.3 性能监控
添加调用统计和性能监控：
python复制@app.middleware("http")
async def monitor_requests(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
# 记录指标
metrics.record(
path=request.url.path,
method=request.method,
duration=process_time
)
return response
8. 项目总结与心得
通过这个项目的实践，我总结了以下几点关键经验：
-
原子化设计是构建可维护AI系统的关键。将复杂功能拆解为单一职责的小技能，可以显著提高系统的灵活性和可扩展性。
-
标准化协议消除了不同组件之间的耦合。MCP协议就像智能体世界的"通用语言"，使不同来源的技能可以无缝协作。
-
解耦架构带来了部署的灵活性。服务端和客户端的分离使我们能够独立扩展和优化每个部分。
-
防御性编程对AI系统尤为重要。对所有外部依赖（API、模型输出等）都应当做最坏的假设，并做好相应的错误处理和回退机制。
在实际应用中，这套架构已经证明了其价值。我们仅用2周时间就接入了10多个新技能，而客户端代码几乎不需要修改。这种扩展效率在传统的单体架构中是难以想象的。