从0到1实现AI Agent Skills + MCP协议：构建标准化智能体技能体系
本文介绍了基于MCP协议的Agent Skills技术实现方案，通过原子化技能设计和标准化协议，构建可复用的智能体功能单元。系统采用"技能服务端+Agent客户端"架构，支持技能注册、发现和标准化调用。核心内容包括：1) Agent Skills的原子化、标准化设计特点；2) MCP协议解决技能服务化、调用标准化和生态兼容性问题；3) 开源项目实现，涵盖天气查询、文本摘要等示例
·
一、背景与核心概念
随着AI Agent技术的快速发展，智能体的"技能化"已成为核心设计思路。Agent Skills（智能体技能）是将AI Agent能力拆解为原子化、可复用、可管控的功能单元；而MCP（Model Control Protocol）则是让这些技能实现标准化调用的通用协议，二者结合是构建工业级AI Agent的基础。
本文将基于实际落地案例，从核心概念拆解到代码实现，完整讲解如何构建一套支持MCP协议的Agent Skills体系，所有代码已开源至：https://gitee.com/impl/learn-agent-skills。
二、核心设计理念
2.1 Agent Skills的四大核心特点
Agent Skills并非简单的函数封装，而是遵循以下通用设计理念：
- 原子化设计：每个技能仅负责单一功能（如查天气、文本摘要、读文件），不可拆分，便于复用和维护；
- 标准化输入输出：所有技能统一返回Dict格式，包含成功/失败标识、业务数据、提示信息，降低调用端适配成本；
- 解耦设计：技能实现与调用逻辑分离，技能层仅关注"怎么做"，调用层仅关注"调哪个"；
- 可注册/可发现/可管控：技能支持中心化注册、动态发现，便于统一管理和权限控制。
2.2 MCP协议的核心价值
MCP（Model Control Protocol）是智能体技能调用的通用协议标准，核心解决三大问题：
- 技能服务化：将零散的技能封装为标准化服务，支持跨进程/跨语言调用；
- 调用标准化：定义统一的技能元数据、调用请求/响应格式，兼容主流大模型生态；
- 生态兼容性：让不同厂商的Agent都能通过统一协议调用技能，打破生态壁垒。
三、技术架构设计
3.1 整体架构（服务端/客户端分离）
采用经典的"技能服务端 + Agent客户端"架构，贴合生产环境落地需求，项目完整代码可参考https://gitee.com/impl/learn-agent-skills。
┌─────────────────────┐ ┌─────────────────────┐
│ MCP技能服务端 │ │ MCP Agent客户端 │
│ （FastAPI实现） │ │ （DeepSeek大模型） │
│ 1. 技能注册 │ │ 1. LLM决策选技能 │
│ 2. 技能发现 │◄────►│ 2. MCP协议调技能 │
│ 3. 标准化技能调用 │ │ 3. 返回执行结果 │
└─────────────────────┘ └─────────────────────┘
│
▼
┌─────────────────────┐
│ 原子化技能层 │
│ （查天气/摘要/读文件）│
└─────────────────────┘
3.2 目录结构设计
本项目遵循模块化、高内聚低耦合的设计原则，目录结构清晰可扩展，完整目录可在https://gitee.com/impl/learn-agent-skills查看：
learn-agent-skills/
├── skills/ # 原子化技能层
│ ├── base_skills.py # 核心技能实现（查天气/摘要/读文件）
│ └── skill_registry.py # 技能注册中心
├── mcp_server/ # MCP技能服务端
│ ├── mcp_protocol.py # MCP协议定义
│ └── main_server.py # HTTP服务实现（技能注册/发现/调用）
├── mcp_agent/ # MCP Agent客户端
│ ├── mcp_protocol.py # MCP协议定义（与服务端一致）
│ └── main_agent.py # 大模型决策+MCP调用逻辑
├── .env.template # 环境变量模板（需复制为.env配置实际密钥）
├── pyproject.toml # 项目依赖配置
├── uv.lock # 依赖锁定文件
└── test.txt # 测试文件（供文件读取技能使用）
四、核心代码实现
本项目所有核心代码均已开源至https://gitee.com/impl/learn-agent-skills，可直接拉取运行，以下为核心模块的实现解析。
4.1 原子化技能实现（skills/base_skills.py）
以天气查询技能为例，体现原子化、标准化设计，所有技能均遵循统一的输入输出规范：
import requests
import os
from typing import Dict
def get_weather(city: str) -> Dict[str, str]:
"""
原子化技能：仅负责查询指定城市天气
入参标准化：仅接收city参数
出参标准化：统一返回Dict格式（成功/失败标识+业务数据）
"""
try:
appid = os.getenv("WEATHER_APPID")
appsecret = os.getenv("WEATHER_APPSECRET")
if not appid or not appsecret:
return {"error": "未配置天气API密钥"}
api_url = (
f"http://v0.yiketianqi.com/api?unescape=1&version=v91"
f"&appid={appid}&appsecret={appsecret}&city={city}"
)
response = requests.get(api_url, timeout=10)
response.raise_for_status()
data = response.json()
if "city" in data and "data" in data and len(data["data"]) > 0:
return {
"city": city,
"temperature": data["data"][0]["tem"],
"weather": data["data"][0]["wea"],
"message": "查询成功"
}
else:
return {"error": "查询失败：返回数据结构异常"}
except Exception as e:
return {"error": f"天气查询异常：{str(e)}"}
# 文本摘要技能（原子化+标准化）
def summarize_text(text: str, max_length: int = 100) -> Dict[str, str]:
try:
if not text:
return {"error": "文本不能为空"}
summary = text[:max_length].strip() + "..." if len(text) > max_length else text
return {
"summary": summary,
"original_length": len(text),
"summary_length": len(summary),
"message": "摘要成功"
}
except Exception as e:
return {"error": f"文本摘要异常：{str(e)}"}
# 文件读取技能（原子化+标准化）
def read_file_content(file_path: str) -> Dict[str, str]:
try:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
return {
"file_path": file_path,
"content": content,
"message": "文件读取成功"
}
except Exception as e:
return {"error": f"文件读取异常：{str(e)}"}
4.2 MCP协议定义（mcp_server/mcp_protocol.py & mcp_agent/mcp_protocol.py）
遵循行业通用规范，定义标准化的数据结构，服务端与客户端协议保持完全一致，确保通信兼容：
from typing import Dict, List, Any
from typing_extensions import TypedDict
from dataclasses import dataclass
# 技能元数据标准（MCP规范）
class MCPSkillMetadata(TypedDict):
name: str # 技能唯一标识
description: str # 技能描述
parameters: Dict[str, Any] # 参数定义（JSON Schema）
return_type: str # 返回类型
version: str = "1.0.0" # 技能版本
# 技能调用请求标准（MCP规范）
@dataclass
class MCPInvokeRequest:
skill_name: str # 要调用的技能名
parameters: Dict[str, Any] # 技能入参
request_id: str = "" # 请求ID（追踪用）
timeout: int = 30 # 超时时间
# 技能调用响应标准（MCP规范）
@dataclass
class MCPInvokeResponse:
success: bool # 调用是否成功
data: Dict[str, Any] # 响应数据
error: str = "" # 错误信息
request_id: str = "" # 对应请求ID
4.3 MCP技能服务端实现（mcp_server/main_server.py）
基于FastAPI实现标准化的技能服务，提供技能发现和技能调用两大核心接口，是整个技能体系的服务中枢：
import sys
import os
import json
import uuid
from fastapi import FastAPI, HTTPException, Body
import uvicorn
from dotenv import load_dotenv
# 解决模块导入路径问题
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.dirname(current_dir)
sys.path.append(project_root)
from skills.skill_registry import SkillRegistry
from skills.base_skills import get_weather, summarize_text, read_file_content
from .mcp_protocol import MCPSkillMetadata, MCPInvokeRequest, MCPInvokeResponse
# 初始化
load_dotenv()
app = FastAPI(title="MCP技能服务端", version="1.0.0")
# 1. 技能注册
skill_registry = SkillRegistry()
skill_registry.register_skill("get_weather", get_weather)
skill_registry.register_skill("summarize_text", summarize_text)
skill_registry.register_skill("read_file_content", read_file_content)
# 2. MCP技能元数据配置
MCP_SKILL_METADATA = {
"get_weather": MCPSkillMetadata(
name="get_weather",
description="根据城市名称查询实时天气",
parameters={
"type": "object",
"properties": {"city": {"type": "string", "description": "城市名"}},
"required": ["city"]
},
return_type="object"
),
"summarize_text": MCPSkillMetadata(
name="summarize_text",
description="文本摘要",
parameters={
"type": "object",
"properties": {
"text": {"type": "string"},
"max_length": {"type": "integer", "default": 100}
},
"required": ["text"]
},
return_type="object"
),
"read_file_content": MCPSkillMetadata(
name="read_file_content",
description="读取本地文件",
parameters={
"type": "object",
"properties": {"file_path": {"type": "string"}},
"required": ["file_path"]
},
return_type="object"
)
}
# 3. MCP核心接口
@app.get("/mcp/skills")
async def list_mcp_skills():
"""MCP接口：技能发现"""
skills = list(MCP_SKILL_METADATA.values())
return {"skills": skills, "total": len(skills)}
@app.post("/mcp/invoke")
async def invoke_mcp_skill(request: MCPInvokeRequest = Body(...)):
"""MCP接口：技能调用"""
try:
if request.skill_name not in skill_registry.get_all_skills():
raise HTTPException(status_code=404, detail=f"技能 {request.skill_name} 不存在")
skill_func = skill_registry.get_all_skills()[request.skill_name]
result = skill_func(**request.parameters)
return MCPInvokeResponse(
success=True,
data=result,
request_id=request.request_id or str(uuid.uuid4())
).__dict__
except Exception as e:
return MCPInvokeResponse(
success=False,
data={},
error=str(e),
request_id=request.request_id or str(uuid.uuid4())
).__dict__
# 启动服务
if __name__ == "__main__":
print("===== MCP技能服务端启动 =====")
print("服务地址：http://127.0.0.1:8000")
uvicorn.run(app, host="127.0.0.1", port=8000)
4.4 MCP Agent客户端实现（mcp_agent/main_agent.py）
结合大模型决策与MCP协议调用，实现"用户提问→大模型选技能→MCP调技能→返回结果"的完整流程：
import sys
import os
import json
import requests
import uuid
from dotenv import load_dotenv
# 解决模块导入路径问题
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.dirname(current_dir)
sys.path.append(project_root)
from .mcp_protocol import MCPInvokeRequest, MCPInvokeResponse
# 初始化
load_dotenv()
MCP_SERVER_URL = "http://127.0.0.1:8000"
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
# MCP客户端工具类
class MCPClient:
def __init__(self, server_url: str):
self.server_url = server_url
def list_skills(self) -> dict:
"""调用MCP服务端：获取技能列表"""
try:
response = requests.get(f"{self.server_url}/mcp/skills", timeout=10)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"MCP技能发现失败：{str(e)}")
return {"skills": [], "total": 0}
def invoke_skill(self, skill_name: str, parameters: dict) -> MCPInvokeResponse:
"""调用MCP服务端：执行技能"""
request = MCPInvokeRequest(
skill_name=skill_name,
parameters=parameters,
request_id=str(uuid.uuid4())
)
try:
response = requests.post(
f"{self.server_url}/mcp/invoke",
json=request.__dict__,
timeout=30
)
response.raise_for_status()
result = response.json()
return MCPInvokeResponse(
success=result["success"],
data=result["data"],
error=result["error"],
request_id=result["request_id"]
)
except Exception as e:
return MCPInvokeResponse(
success=False,
data={},
error=f"MCP调用失败：{str(e)}",
request_id=request.request_id
)
# MCP Agent核心逻辑
class MCPAgent:
def __init__(self, mcp_client: MCPClient):
self.mcp_client = mcp_client
def get_llm_decision(self, user_query: str) -> dict:
"""大模型决策：选择要调用的MCP技能"""
if not DEEPSEEK_API_KEY:
print("错误：未配置DEEPSEEK_API_KEY")
return {}
# 获取MCP技能列表（供大模型参考）
mcp_skills = self.mcp_client.list_skills()
skills_prompt = "\n".join([
f"- {skill['name']}：{skill['description']}，参数：{json.dumps(skill['parameters'])}"
for skill in mcp_skills["skills"]
])
headers = {
"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": f"""你是MCP客户端Agent，可调用以下技能：
{skills_prompt}
仅输出JSON：{{"skill_name": "技能名", "parameters": {{参数}}}}，无其他内容。
无合适技能输出：{{"skill_name": "", "parameters": {{}}}}"""
},
{"role": "user", "content": user_query}
],
"temperature": 0.0
}
try:
response = requests.post(DEEPSEEK_API_URL, headers=headers, json=payload, timeout=30)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"].strip()
content = content.replace("```json", "").replace("```", "").strip()
return json.loads(content)
except Exception as e:
print(f"LLM决策失败：{str(e)}")
return {}
def run(self, user_query: str):
"""Agent主运行逻辑"""
print(f"\n===== MCP Agent客户端 =====")
print(f"用户提问：{user_query}")
# 1. 大模型决策选技能
decision = self.get_llm_decision(user_query)
if not decision or not decision.get("skill_name"):
return {"error": "未匹配到MCP技能"}
print(f"LLM决策：调用 {decision['skill_name']}，参数 {decision['parameters']}")
# 2. MCP协议调用技能
response = self.mcp_client.invoke_skill(
skill_name=decision["skill_name"],
parameters=decision["parameters"]
)
# 3. 返回结果
print(f"MCP调用结果：成功={response.success}")
return {
"success": response.success,
"data": response.data,
"error": response.error
}
# 启动Agent
if __name__ == "__main__":
mcp_client = MCPClient(MCP_SERVER_URL)
agent = MCPAgent(mcp_client)
# 测试用例
print("===== 测试MCP Agent客户端 =====")
result1 = agent.run("查上海今天的天气")
print("测试1结果：", result1)
result2 = agent.run("读取当前目录下test.txt的内容")
print("测试2结果：", result2)
五、部署与运行
5.1 环境准备
- 拉取项目代码：
git clone https://gitee.com/impl/learn-agent-skills.git
cd learn-agent-skills
- 安装依赖：本项目使用uv进行依赖管理，执行以下命令完成依赖安装：
uv sync
- 配置环境变量：复制环境变量模板并配置实际密钥，项目中提供了
.env.template模板文件：
# 复制模板为.env文件
cp .env.template .env
# 编辑.env文件，配置实际的API密钥
# 天气API配置
WEATHER_APPID=你的天气API_APPID
WEATHER_APPSECRET=你的天气API_APPSECRET
# DeepSeek配置
DEEPSEEK_API_KEY=你的DeepSeek_API_KEY
5.2 启动MCP技能服务端
# Windows PowerShell
$env:PYTHONPATH = "你的项目根目录（learn-agent-skills）"
uv run .\mcp_server\main_server.py
启动成功后，服务端将运行在http://127.0.0.1:8000，提供MCP协议的技能发现和调用接口。
5.3 启动MCP Agent客户端
新建终端，执行以下命令启动客户端，客户端将连接本地的MCP技能服务端并执行测试用例：
# Windows PowerShell
$env:PYTHONPATH = "你的项目根目录（learn-agent-skills）"
uv run .\mcp_agent\main_agent.py
六、关键问题与解决方案
在项目开发和运行过程中，我们解决了多个Python开发和AI Agent开发的常见问题，相关解决方案已整合到https://gitee.com/impl/learn-agent-skills的项目代码中，供大家参考。
6.1 模块导入路径问题
Python解释器默认仅搜索当前目录，跨模块导入时会出现ModuleNotFoundError，解决方案是手动将项目根目录加入Python搜索路径：
import sys
import os
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.dirname(current_dir)
sys.path.append(project_root)
6.2 Python版本兼容问题
Python 3.12以下版本对TypedDict的支持存在兼容问题，需使用typing_extensions.TypedDict替代typing.TypedDict，并在pyproject.toml中添加对应依赖：
dependencies = [
"typing-extensions>=4.8.0"
]
6.3 第三方API适配问题
不同第三方API的返回结构存在差异，避免依赖固定字段（如原天气API的errcode字段），采用通用的字段检查方式提升代码兼容性：
# 原逻辑（依赖固定errcode字段）
if data.get("errcode") == 0:
# 新逻辑（通用字段检查，适配不同API返回结构）
if "city" in data and "data" in data and len(data["data"]) > 0:
七、扩展与落地建议
本项目https://gitee.com/impl/learn-agent-skills为基础的MCP+Agent Skills实现，可基于此进行灵活扩展，适配不同的业务场景和生产环境需求：
- 技能扩展：新增技能只需实现原子化函数，并在MCP服务端注册元数据，无需修改Agent核心逻辑，支持技能的热插拔；
- 权限控制：在MCP服务端添加技能调用权限校验模块，支持按角色、用户管控技能访问权限，适配企业级权限管理需求；
- 监控运维：增加技能调用日志、耗时统计、失败重试、熔断降级机制，提升服务稳定性，便于线上问题排查；
- 多模型兼容：Agent客户端的LLM决策模块与MCP调用模块解耦，可快速替换为GPT、文心一言、通义千问等其他大模型；
- 分布式部署：将MCP服务端部署为微服务，配合Nginx做负载均衡，支持多Agent节点共享技能池，实现大规模部署；
- 技能中台化：基于本项目扩展技能注册中心、技能版本管理、技能市场等功能，构建企业级AI Agent技能中台。
八、总结
本文基于开源项目https://gitee.com/impl/learn-agent-skills，从核心概念到代码实现，完整讲解了Agent Skills + MCP协议的落地实现方案，打造了一套标准化、可扩展、高兼容的智能体技能体系。
本项目的核心收获包括：
- Agent Skills的核心是原子化、标准化、解耦设计，这是智能体能力复用、快速扩展的基础；
- MCP协议是技能标准化调用的关键，实现了技能的服务化封装与跨生态、跨平台兼容；
- 服务端/客户端分离的架构设计，贴合生产环境落地需求，便于模块独立升级、扩展和维护；
- 解决了Python跨模块导入、版本兼容、第三方API适配等实际开发问题，提供了可直接复用的解决方案。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
11
0- 0
已为社区贡献8条内容
所有评论(0)