CLI + MCP + Skill:2026年AI Agent开发的三大范式详解与实战
引言:AI Agent开发范式的演进
2026年,AI Agent的开发已从早期的单一模型调用,演进为模块化、标准化和工具化的系统工程。在这一演进过程中,三种核心范式脱颖而出,共同构成了现代AI Agent开发的基石:
- CLI(命令行界面):作为人机交互与自动化执行的基础入口。
- MCP(模型上下文协议):作为连接AI模型与外部工具/数据的标准化通信层。
- Skill(技能):作为封装可复用业务逻辑与工具调用的功能单元。
本文将深入剖析这三大范式,阐明其各自角色、交互关系,并通过一个完整的“智能数据分析助手”Agent实战项目,展示如何将它们有机结合,构建强大、可维护的AI应用。
范式一:CLI - 统一的控制面与执行入口
CLI是开发者、用户乃至其他系统与AI Agent交互的首要界面。它超越了传统命令执行,成为Agent的“总控台”。
核心价值
- 标准化入口:为Agent的所有功能(对话、任务执行、技能调用)提供一致的启动和配置方式。
- 自动化集成:易于嵌入CI/CD流水线、cron作业或其他自动化脚本。
- 环境与配置管理:集中处理环境变量、认证信息、模型选择等配置。
实战:构建一个Agent CLI骨架
我们使用 argparse (Python标准库) 或 typer (更现代) 来创建CLI。以下使用 typer 为例:
# 首先安装依赖
pip install typer rich
# agent_cli.py
import typer
from typing import Optional
import json
from rich.console import Console
from rich.table import Table
from rich import print as rprint
app = typer.Typer(help="🤖 智能数据分析助手 CLI")
console = Console()
# 模拟的Agent核心和技能库
agent_core = {"status": "ready"}
skills_registry = ["data_fetcher", "chart_generator", "report_writer"]
@app.command()
def start(
task: str = typer.Argument(..., help="启动Agent执行的具体任务描述"),
model: str = typer.Option("gpt-4", "--model", "-m", help="选择使用的AI模型"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="显示详细日志")
):
"""
启动AI Agent执行一项任务。
"""
rprint(f"[bold green]🚀 启动Agent...[/bold green]")
rprint(f" 任务: {task}")
rprint(f" 模型: {model}")
rprint(f" 详细模式: {verbose}")
# 此处会触发后续的MCP调用和Skill执行
# 例如:process_task_via_mcp(task, model)
rprint("[bold green]✅ 任务已接收并进入处理队列。[/bold green]")
@app.command()
def list_skills():
"""列出当前Agent已注册的所有技能。"""
table = Table(title="🛠️ 已注册技能")
table.add_column("技能名称", style="cyan")
table.add_column("描述", style="magenta")
table.add_column("状态", style="green")
# 模拟从技能库获取信息
for skill in skills_registry:
table.add_row(skill, f"执行与'{skill}'相关的操作", "✅ 可用")
console.print(table)
@app.command()
def status():
"""查看Agent核心状态。"""
rprint(json.dumps(agent_core, indent=2))
if __name__ == "__main__":
app()
使用这个CLI:
# 查看帮助
python agent_cli.py --help
# 启动一个任务
python agent_cli.py start "分析本月销售数据并生成趋势图" --model claude-3-5-sonnet
# 列出技能
python agent_cli.py list-skills
# 查看状态
python agent_cli.py status
CLI范式要点:CLI不包含核心业务逻辑,它只是一个调度器和配置器,其命令最终会转化为对MCP层的调用。
范式二:MCP - 模型上下文的标准化协议
MCP (Model Context Protocol) 是一种开放协议,用于在AI模型(如ChatGPT、Claude)与工具(Tools)、数据源(Data Sources) 之间建立标准化、声明式的连接。它解决了模型如何安全、可控地访问外部能力的问题。
核心概念
- Server(服务器):提供工具或数据访问能力的后端服务。一个Agent可以连接多个MCP Server。
- Tools(工具):Server暴露的可调用函数,模型可以决定在何时、传入何种参数来调用它们。
- Resources(资源):Server提供的可读数据源(如数据库表、文件列表)。
- Prompts(提示词):可复用的提示词模板。
实战:为Agent实现一个MCP Server(数据获取工具)
假设我们有一个“数据获取”MCP Server,它向AI模型暴露一个 query_sales_data 工具。
# mcp_data_server.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.server import Server
from mcp.server.models import Tool
import pandas as pd
from datetime import datetime, timedelta
import json
# 创建MCP Server实例
server = Server("sales-data-server")
# 1. 声明Server提供的工具
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
return [
Tool(
name="query_sales_data",
description="查询指定日期范围内的销售数据。",
inputSchema={
"type": "object",
"properties": {
"start_date": {"type": "string", "description": "开始日期 (YYYY-MM-DD)"},
"end_date": {"type": "string", "description": "结束日期 (YYYY-MM-DD)"},
"product_category": {"type": "string", "description": "产品类别(可选)", "default": "all"}
},
"required": ["start_date", "end_date"]
}
)
]
# 2. 实现工具的执行逻辑
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> dict:
if name == "query_sales_data":
return await execute_query_sales_data(**arguments)
raise ValueError(f"未知工具: {name}")
async def execute_query_sales_data(start_date: str, end_date: str, product_category: str = "all") -> dict:
"""模拟查询销售数据。"""
# 在实际应用中,这里会连接数据库
# 此处返回模拟数据
dates = pd.date_range(start_date, end_date)
data = []
for date in dates:
for category in ['电子产品', '家居用品', '图书']:
if product_category != 'all' and category != product_category:
continue
revenue = 1000 + (hash(f"{date}{category}") % 500) # 模拟收入
data.append({"date": date.strftime("%Y-%m-%d"), "category": category, "revenue": revenue})
df = pd.DataFrame(data)
summary = df.groupby('category')['revenue'].sum().to_dict()
return {
"status": "success",
"message": f"查询到 {len(df)} 条记录。",
"data_preview": df.head().to_dict('records'), # 预览前5条
"summary": summary,
"metadata": {
"query_params": {"start_date": start_date, "end_date": end_date, "product_category": product_category}
}
}
# 3. 运行Server(通常通过stdio与模型客户端通信)
async def main():
async with server.run_stdio_server() as (read_stream, write_stream):
# 这里Server会持续运行,等待模型客户端的指令
await asyncio.Future() # 永久运行
if __name__ == "__main__":
asyncio.run(main())
MCP范式要点:MCP Server是能力的提供者,它通过标准协议告诉模型“我能做什么”(list_tools),并在模型请求时“执行具体操作”(call_tool)。Agent的核心(或模型)作为MCP Client,会动态发现并调用这些工具。
范式三:Skill - 模块化与可复用的功能单元
Skill是对一个或多个相关工具调用、决策逻辑和提示词工程的高层封装。它代表Agent能完成的一个完整“任务”或“能力”。
核心特征
- 声明式定义:描述Skill的输入、输出、所需工具和前置条件。
- 可组合性:简单的Skill可以组合成复杂的Skill(例如,“获取数据” + “生成图表” = “生成数据报告”)。
- 独立可测试:每个Skill都可以在脱离主Agent的情况下进行单元测试。
实战:定义并实现一个“生成销售图表”Skill
# skills/chart_generator_skill.py
from typing import TypedDict, Optional
import plotly.graph_objects as go
import plotly.io as pio
import json
import os
class ChartInput(TypedDict):
"""生成图表Skill的输入参数。"""
data: list # 数据列表,通常来自`query_sales_data`工具的结果
chart_type: str # 图表类型,如 'line', 'bar', 'pie'
title: str
output_format: str # 'html', 'png', 'json'
class ChartOutput(TypedDict):
"""生成图表Skill的输出结果。"""
status: str
chart_path: Optional[str] # 生成的图表文件路径
chart_html: Optional[str] # 如果是HTML格式,直接返回内容
error_message: Optional[str]
class ChartGeneratorSkill:
"""技能:根据数据生成可视化图表。"""
name = "generate_sales_chart"
description = "将销售数据转换为可视化图表。"
required_tools = ["query_sales_data"] # 此技能依赖的工具
@classmethod
def execute(cls, input_params: ChartInput, context: dict) -> ChartOutput:
"""
执行技能。
:param input_params: 技能输入参数。
:param context: 执行上下文,包含可用的工具调用函数等。
:return: 技能执行结果。
"""
try:
data = input_params['data']
chart_type = input_params['chart_type']
title = input_params['title']
# 1. 数据转换 (根据实际数据结构调整)
# 假设data是 `query_sales_data` 返回的 `data_preview` 格式
df_data = pd.DataFrame(data)
if df_data.empty:
return {"status": "error", "error_message": "输入数据为空。"}
# 2. 根据图表类型创建Plotly图形
fig = None
if chart_type == 'line':
fig = go.Figure()
for category in df_data['category'].unique():
cat_data = df_data[df_data['category'] == category]
fig.add_trace(go.Scatter(x=cat_data['date'], y=cat_data['revenue'],
mode='lines+markers', name=category))
fig.update_layout(title=title, xaxis_title='日期', yaxis_title='销售额')
elif chart_type == 'bar':
# 按类别聚合
summary = df_data.groupby('category')['revenue'].sum().reset_index()
fig = go.Figure([go.Bar(x=summary['category'], y=summary['revenue'])])
fig.update_layout(title=title, xaxis_title='产品类别', yaxis_title='总销售额')
else:
return {"status": "error", "error_message": f"不支持的图表类型: {chart_type}"}
# 3. 输出处理
output_format = input_params.get('output_format', 'html')
output_dir = "output/charts"
os.makedirs(output_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"sales_chart_{timestamp}"
if output_format == 'html':
html_path = os.path.join(output_dir, f"{filename}.html")
fig.write_html(html_path)
with open(html_path, 'r', encoding='utf-8') as f:
html_content = f.read()
return {
"status": "success",
"chart_path": html_path,
"chart_html": html_content
}
elif output_format == 'png':
img_path = os.path.join(output_dir, f"{filename}.png")
fig.write_image(img_path)
return {"status": "success", "chart_path": img_path}
else:
return {"status": "error", "error_message": f"不支持的输出格式: {output_format}"}
except Exception as e:
return {"status": "error", "error_message": f"生成图表时出错: {str(e)}"}
# 技能的使用示例
if __name__ == "__main__":
# 模拟从MCP工具调用获取的数据
mock_data = [
{"date": "2026-07-01", "category": "电子产品", "revenue": 1200},
{"date": "2026-07-01", "category": "家居用品", "revenue": 800},
{"date": "2026-07-02", "category": "电子产品", "revenue": 1500},
{"date": "2026-07-02", "category": "家居用品", "revenue": 950},
]
skill_input: ChartInput = {
"data": mock_data,
"chart_type": "line",
"title": "近两日销售趋势",
"output_format": "html"
}
skill = ChartGeneratorSkill()
result = skill.execute(skill_input, context={})
print(json.dumps(result, indent=2, ensure_ascii=False))
Skill范式要点:Skill是业务的封装。它知道完成一个特定任务需要调用哪些工具(通过MCP),如何处理这些工具的返回结果,以及如何将结果转化为最终输出。Skill使Agent的能力变得可插拔、可管理。
综合实战:构建“智能数据分析助手”Agent
现在,我们将三大范式串联起来,构建一个完整的Agent。
项目结构
smart_data_agent/
├── agent_cli.py # CLI入口 (范式一)
├── mcp_servers/ # MCP Server集合 (范式二)
│ ├── data_server.py # 数据查询Server
│ └── chart_server.py # 图表生成Server (可独立,也可作为Skill内部调用)
├── skills/ # Skill集合 (范式三)
│ ├── chart_generator_skill.py
│ └── report_skill.py # (可扩展)组合数据获取和图表生成的报告Skill
├── agent_core.py # Agent核心,协调CLI、MCP Client和Skill
└── requirements.txt
Agent核心协调逻辑 (agent_core.py 简化版)
# agent_core.py
import asyncio
from typing import List, Dict, Any
from mcp import ClientSession, StdioServerParameters
import importlib
import sys
import os
class SmartDataAgent:
def __init__(self):
self.mcp_clients: Dict[str, ClientSession] = {}
self.skills_registry: Dict[str, Any] = {}
self.load_skills()
def load_skills(self):
"""动态加载skills目录下的所有技能。"""
skills_dir = "skills"
for filename in os.listdir(skills_dir):
if filename.endswith("_skill.py"):
module_name = f"skills.{filename[:-3]}"
spec = importlib.util.spec_from_file_location(module_name, os.path.join(skills_dir, filename))
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
# 假设每个skill模块都有一个`SkillClass`类
for attr_name in dir(module):
attr = getattr(module, attr_name)
if hasattr(attr, 'name') and hasattr(attr, 'execute'):
self.skills_registry[attr.name] = attr
print(f"✅ 已加载技能: {attr.name}")
async def connect_to_mcp_server(self, server_command: List[str]):
"""连接到一个MCP Server。"""
params = StdioServerParameters(command=server_command[0], args=server_command[1:])
session = ClientSession(params)
await session.__aenter__() # 简化连接
# 初始化会话,获取Server提供的工具列表
await session.initialize()
tools = await session.list_tools()
server_name = server_command[0]
self.mcp_clients[server_name] = session
print(f"🔗 已连接MCP Server: {server_name}, 可用工具: {[t.name for t in tools.tools]}")
return session
async def execute_skill(self, skill_name: str, skill_input: Dict, context: Dict) -> Dict:
"""执行一个指定的技能。"""
if skill_name not in self.skills_registry:
return {"status": "error", "message": f"技能 '{skill_name}' 未找到。"}
skill_class = self.skills_registry[skill_name]
# 技能执行可能需要调用MCP工具,这里需要将MCP client传递给技能上下文
context['mcp_clients'] = self.mcp_clients
result = skill_class.execute(skill_input, context)
return result
async def process_task(self, task_description: st
更多推荐

所有评论(0)