一、MCP是什么?
Model Context Protocol(MCP)是由人工智能公司Anthropic于2024年推出的开放协议,它像一座桥梁,将大语言模型(LLM)与外部数据源、工具无缝连接。简单来说,MCP让AI助手不仅能“思考”,还能“动手操作”——无论是读取本地文件、调用API,还是控制智能设备,都变得像插拔USB一样简单。
1. 三大核心组件
MCP主机:AI应用本体(如Claude桌面端)
MCP客户端:内嵌在主机中的协议接口
MCP服务器:轻量级程序,负责连接具体数据源或工具
协议层:定义资源(Resources)、工具(Tools)、提示(Prompts)的交互规则
资源:类似REST API的GET请求(如读取文件) 工具:可执行函数(需用户授权,如发送邮件) 提示:预置任务模板(如生成SQL语句)
传输层:支持HTTP/SSE通信,采用JSON-RPC 2.0标准
AI通过MCP客户端发现天气服务器 调用get_weather工具并传入地理位置参数 服务器从气象API获取实时数据并返回 AI生成自然语言回答
开发效率提升
预置TypeScript/Python SDK,30行代码即可创建服务器
#连接SQLite数据库仅需定义查询函数+启动HTTP服务import sqlite3
import sqlite3
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs
class SecureMCPService(BaseHTTPRequestHandler):
"""增强版MCP协议服务端,支持参数化查询和错误处理"""
def _set_headers(self, status_code=200):
"""设置HTTP响应头"""
self.send_response(status_code)
self.send_header('Content-type', 'application/json')
self.end_headers()
def _parse_request(self):
"""解析请求参数"""
content_length = int(self.headers['Content-Length'])
raw_data = self.rfile.read(content_length).decode()
try:
return json.loads(raw_data)
except json.JSONDecodeError:
return None
def _execute_safe_query(self, query, params=None):
"""安全执行数据库查询"""
try:
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
cursor.execute(query, params or ())
if query.strip().upper().startswith('SELECT'):
columns = [col[0] for col in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
else:
conn.commit()
return {"affected_rows": cursor.rowcount}
except sqlite3.Error as e:
return {"error": f"Database error: {str(e)}"}
def do_POST(self):
"""处理POST请求"""
parsed_path = urlparse(self.path)
# 路由处理
if parsed_path.path == '/api/query':
request_data = self._parse_request()
if not request_data or 'statement' not in request_data:
self._set_headers(400)
self.wfile.write(json.dumps({"error": "Invalid request"}).encode())
return
# 执行安全查询
result = self._execute_safe_query(
request_data['statement'],
request_data.get('parameters')
)
self._set_headers()
self.wfile.write(json.dumps(result).encode())
else:
self._set_headers(404)
self.wfile.write(json.dumps({"error": "Endpoint not found"}).encode())
def start_server(port=8080):
"""启动安全HTTP服务器"""
server_address = ('', port)
httpd = HTTPServer(server_address, SecureMCPService)
print(f'MCP服务运行中,监听端口:{port}')
try:
httpd.serve_forever()
except KeyboardInterrupt:
httpd.server_close()
print('n服务已停止')
if __name__ == '__main__':
# 初始化示例数据库
with sqlite3.connect('example.db') as conn:
conn.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY, name TEXT, email TEXT)''')
conn.commit()
start_server()
动态交互能力
AI可实时发现新接入的服务器(如插入U盘自动加载文件服务)
安全沙箱机制
所有工具调用需用户授权
凭据隔离存储,避免敏感数据泄露
混合部署支持
本地数据(如企业内网数据库)+云端服务(如GitHub API)统一接入
开源生态繁荣
Anthropic提供预构建服务器(Google Drive、Slack等)
社区贡献仓库已有200+工具集成
跨平台兼容性
支持从IDE插件到工业控制面板的全场景覆盖
开发者工具 IDE实时读取项目文件,智能补全代码 自动生成SQL语句并直接执行(需授权) 企业服务 客服机器人调取CRM数据提供个性化回复 会议纪要自动同步到Notion/钉钉 物联网 语音控制智能家居(通过MCP连接HomeAssistant) 工厂设备状态实时监控与预警 医疗健康 AI助手安全读取电子病历(符合HIPAA合规性) 科研文献自动检索与摘要生成
更细粒度的权限控制(如基于角色的访问) 边缘计算场景优化(低带宽环境下的性能提升) 与RPA工具的深度融合(自动化工作流编排)
长
按
关
注
网络安全社团公众号
微信号 : qlnu_ctf
新浪微博:齐鲁师范学院网络安全社团
推荐站内搜索:最好用的开发软件、免费开源系统、渗透测试工具云盘下载、最新渗透测试资料、最新黑客工具下载……
还没有评论,来说两句吧...