StructBERT中文情感模型API安全加固：添加JWT认证接入企业内网

Ne0inhk

— 17 min read

StructBERT中文情感模型API安全加固：添加JWT认证接入企业内网

1. 引言

如果你在企业内部部署了一个AI模型服务，比如这个StructBERT中文情感分析模型，你可能会遇到一个很实际的问题：怎么保证只有公司内部的系统能调用这个API，而外部的人无法访问？直接暴露在公网上的API端口，就像把家门钥匙放在门口的地垫下面，谁都能找到。

我最近帮一个客户部署了StructBERT情感分析服务，他们需要把这个服务集成到自己的CRM系统中，用于分析客户反馈的情绪。最初的版本很简单，就是启动服务，然后通过8080端口直接调用。但他们的安全团队提出了明确要求：必须要有身份验证机制，不能谁都能调用。

这就是我们今天要解决的问题——如何给StructBERT的API服务加上JWT（JSON Web Token）认证，让它能够安全地接入企业内网。我会带你一步步实现这个功能，从理解JWT是什么，到具体怎么修改代码，再到怎么在企业环境中使用。

2. 为什么需要API安全加固？

2.1 企业环境的安全需求

在企业内部，AI模型服务通常不是孤立存在的。它需要和其他系统集成，比如：

  • 客户关系管理系统（CRM）：分析客户反馈的情绪倾向
  • 客服系统：实时评估客服对话中的客户情绪
  • 社交媒体监控平台：分析品牌相关的评论情感
  • 产品反馈系统：自动分类用户评价的正负面

这些系统都需要调用情感分析API，但你不能让每个系统都无限制地访问。你需要：

  1. 身份验证：确认调用方是谁
  2. 访问控制：控制谁能调用什么接口
  3. 使用统计：记录谁在什么时候调用了多少次
  4. 防止滥用：避免外部恶意调用消耗资源

2.2 JWT认证的优势

JWT是一种流行的认证方案，它有几个明显的优点：

  • 无状态：服务器不需要保存会话信息，所有必要信息都在token里
  • 可扩展：可以在token里携带用户角色、权限等额外信息
  • 标准化：有明确的标准规范，各种编程语言都有现成库支持
  • 适合API：特别适合RESTful API的认证场景

相比于传统的session-cookie方案或者简单的API key，JWT更适合微服务架构下的API认证。

3. 理解JWT的工作原理

3.1 JWT是什么？

简单来说，JWT就是一个加密的字符串，里面包含了一些信息。它由三部分组成，用点号分隔：

头部.载荷.签名
  • 头部（Header）：说明token的类型和使用的加密算法
  • 载荷（Payload）：存放实际的数据，比如用户ID、过期时间等
  • 签名（Signature）：用密钥对前两部分进行签名，防止被篡改

3.2 JWT的工作流程

让我用一个实际的例子来说明JWT是怎么工作的：

  1. 用户登录：用户提供用户名密码，服务器验证通过
  2. 生成token：服务器生成一个JWT token返回给客户端
  3. 携带token：客户端在后续请求的Header中携带这个token
  4. 验证token：服务器验证token的签名和有效期
  5. 处理请求：如果验证通过，就执行请求的业务逻辑

整个过程不需要服务器保存任何会话状态，所有的验证信息都在token里。

3.3 JWT在企业API中的应用

在企业内部，我们通常这样使用JWT：

  1. 为每个内部系统分配一个密钥：比如CRM系统用一个密钥，客服系统用另一个
  2. 系统启动时获取token：系统用自己的密钥生成token
  3. 调用API时携带token：在Authorization头中带上token
  4. API服务验证token：用对应的公钥验证token的合法性

这样，即使token被截获，攻击者没有对应的私钥也无法伪造新的token。

4. 为StructBERT API添加JWT认证

现在我们来实际操作，给StructBERT的API服务加上JWT认证。我会分步骤讲解，确保你能跟着做出来。

4.1 准备工作

首先，我们需要安装JWT相关的Python库。打开终端，进入你的项目目录：

cd /root/nlp_structbert_sentiment-classification_chinese-base

然后安装必要的依赖：

pip install pyjwt cryptography
  • pyjwt：处理JWT的生成和验证
  • cryptography：提供加密算法支持

4.2 生成RSA密钥对

JWT通常使用非对称加密，也就是有一对公钥和私钥。私钥用来生成token，公钥用来验证token。我们来生成一对密钥：

# generate_keys.py from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import serialization # 生成私钥 private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, ) # 生成公钥 public_key = private_key.public_key() # 保存私钥 with open("private_key.pem", "wb") as f: f.write(private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() )) # 保存公钥 with open("public_key.pem", "wb") as f: f.write(public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo )) print("密钥对已生成：private_key.pem, public_key.pem")

运行这个脚本，你会在当前目录得到两个文件：private_key.pem（私钥）和public_key.pem（公钥）。私钥要妥善保管，不能泄露；公钥可以分发给需要验证token的服务。

4.3 修改API服务代码

现在我们来修改StructBERT的API服务代码。打开原来的main.py文件，我们给它加上JWT认证层。

# app/main_with_jwt.py import jwt import time from functools import wraps from flask import Flask, request, jsonify from cryptography.hazmat.primitives import serialization app = Flask(__name__) # 加载公钥用于验证token with open("public_key.pem", "rb") as key_file: public_key = serialization.load_pem_public_key(key_file.read()) # 这里是你原来的模型加载和预测代码 # 为了简洁，我用伪代码表示 # from your_model import load_model, predict # model = load_model() def token_required(f): """JWT认证装饰器""" @wraps(f) def decorated(*args, **kwargs): token = None # 从请求头获取token if 'Authorization' in request.headers: auth_header = request.headers['Authorization'] if auth_header.startswith('Bearer '): token = auth_header.split(' ')[1] if not token: return jsonify({'error': 'Token is missing'}), 401 try: # 验证token data = jwt.decode( token, public_key, algorithms=['RS256'] ) # 检查token是否过期 if 'exp' in data and data['exp'] < time.time(): return jsonify({'error': 'Token has expired'}), 401 # 你可以在这里添加更多的权限检查 # 比如检查data['role']、data['permissions']等 except jwt.ExpiredSignatureError: return jsonify({'error': 'Token has expired'}), 401 except jwt.InvalidTokenError: return jsonify({'error': 'Invalid token'}), 401 # 如果验证通过，继续执行原来的函数 return f(*args, **kwargs) return decorated @app.route('/health', methods=['GET']) def health_check(): """健康检查接口（不需要认证）""" return jsonify({'status': 'healthy'}) @app.route('/predict', methods=['POST']) @token_required # 加上认证装饰器 def predict_sentiment(): """单文本情感预测（需要认证）""" try: data = request.get_json() text = data.get('text', '') if not text: return jsonify({'error': 'Text is required'}), 400 # 调用原来的预测函数 # result = model.predict(text) # 这里用模拟结果代替 result = { 'text': text, 'sentiment': 'positive', 'confidence': 0.95, 'probabilities': {'positive': 0.95, 'negative': 0.03, 'neutral': 0.02} } return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/batch_predict', methods=['POST']) @token_required # 加上认证装饰器 def batch_predict_sentiment(): """批量情感预测（需要认证）""" try: data = request.get_json() texts = data.get('texts', []) if not texts or not isinstance(texts, list): return jsonify({'error': 'Texts must be a non-empty list'}), 400 # 调用原来的批量预测函数 # results = model.batch_predict(texts) # 这里用模拟结果代替 results = [] for i, text in enumerate(texts): results.append({ 'id': i, 'text': text, 'sentiment': 'positive' if i % 2 == 0 else 'negative', 'confidence': 0.85 + (i * 0.02), 'probabilities': {'positive': 0.7, 'negative': 0.2, 'neutral': 0.1} }) return jsonify({'results': results}) except Exception as e: return jsonify({'error': str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=8080, debug=False)

这个修改主要做了几件事：

  1. 添加了JWT验证装饰器：token_required函数会检查每个请求的token
  2. 健康检查接口不需要认证：这样监控系统可以随时检查服务状态
  3. 预测接口需要认证：只有携带有效token的请求才能调用
  4. 详细的错误信息：告诉调用方token为什么无效（过期、格式错误等）

4.4 创建token生成工具

为了让其他系统能够调用我们的API，我们需要提供一个生成token的工具。这个工具通常由认证服务器提供，这里我们简化一下，直接写一个生成脚本：

# generate_token.py import jwt import time from cryptography.hazmat.primitives import serialization def generate_token(client_id, expires_in=3600): """生成JWT token Args: client_id: 客户端标识，比如 'crm_system' expires_in: token有效期（秒），默认1小时 Returns: JWT token字符串 """ # 加载私钥 with open("private_key.pem", "rb") as key_file: private_key = serialization.load_pem_private_key( key_file.read(), password=None ) # token的有效载荷 payload = { 'client_id': client_id, 'iss': 'structbert_auth_server', # 签发者 'iat': int(time.time()), # 签发时间 'exp': int(time.time()) + expires_in, # 过期时间 'permissions': ['predict', 'batch_predict'] # 权限列表 } # 生成token token = jwt.encode( payload, private_key, algorithm='RS256' ) return token if __name__ == '__main__': # 示例：为CRM系统生成token token = generate_token('crm_system', expires_in=86400) # 24小时有效期 print(f"生成的token: {token}") # 你也可以保存到文件 with open("crm_token.txt", "w") as f: f.write(token) print("Token已保存到 crm_token.txt")

这个脚本可以给不同的内部系统生成不同的token。在实际企业中，你可能会有一个专门的认证服务来管理这些token。

5. 在企业内网中使用

5.1 部署架构

在企业环境中，我们通常这样部署：

企业内网 ├── 认证服务器（生成和验证token） ├── StructBERT API服务（带JWT认证） ├── CRM系统（调用方） ├── 客服系统（调用方） └── 监控系统（健康检查）

5.2 调用示例

现在让我们看看其他系统怎么调用这个带认证的API。这里是一个Python客户端的例子：

# client_example.py import requests import json class StructBERTClient: def __init__(self, base_url, token): self.base_url = base_url self.headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' } def predict(self, text): """单文本情感分析""" url = f"{self.base_url}/predict" data = {'text': text} try: response = requests.post(url, json=data, headers=self.headers) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None def batch_predict(self, texts): """批量情感分析""" url = f"{self.base_url}/batch_predict" data = {'texts': texts} try: response = requests.post(url, json=data, headers=self.headers) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None def health_check(self): """健康检查（不需要token）""" url = f"{self.base_url}/health" try: response = requests.get(url) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"健康检查失败: {e}") return None # 使用示例 if __name__ == '__main__': # 从文件读取token（实际中可能从配置中心获取） with open("crm_token.txt", "r") as f: token = f.read().strip() # 创建客户端 client = StructBERTClient("http://localhost:8080", token) # 检查服务状态 health = client.health_check() print(f"服务状态: {health}") # 单文本分析 result = client.predict("这个产品非常好用，推荐购买！") print(f"单文本分析结果: {json.dumps(result, indent=2, ensure_ascii=False)}") # 批量分析 texts = [ "服务质量很差，不会再来了", "物流速度很快，包装完好", "一般般，没什么特别的感觉" ] batch_result = client.batch_predict(texts) print(f"批量分析结果: {json.dumps(batch_result, indent=2, ensure_ascii=False)}")

5.3 错误处理与重试

在企业环境中，网络可能不稳定，token可能过期，我们需要更健壮的客户端：

# robust_client.py import requests import time from typing import Optional, List, Dict class RobustStructBERTClient: def __init__(self, base_url: str, token: str, max_retries: int = 3): self.base_url = base_url self.token = token self.max_retries = max_retries self.session = requests.Session() # 设置请求头 self.session.headers.update({ 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' }) def _make_request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict]: """带重试的请求方法""" url = f"{self.base_url}/{endpoint}" for attempt in range(self.max_retries): try: response = self.session.request(method, url, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: if response.status_code == 401: # Token过期或无效，需要重新获取 print(f"认证失败: {e}") # 这里可以添加重新获取token的逻辑 return None elif response.status_code == 429: # 请求太频繁，等待后重试 wait_time = 2 ** attempt # 指数退避 print(f"请求限流，等待{wait_time}秒后重试...") time.sleep(wait_time) continue else: print(f"HTTP错误: {e}") return None except requests.exceptions.RequestException as e: print(f"请求异常（尝试{attempt + 1}/{self.max_retries}）: {e}") if attempt < self.max_retries - 1: time.sleep(1) # 等待1秒后重试 else: return None return None def predict_with_retry(self, text: str) -> Optional[Dict]: """带重试的单文本预测""" return self._make_request('POST', 'predict', json={'text': text}) def batch_predict_with_retry(self, texts: List[str]) -> Optional[Dict]: """带重试的批量预测""" return self._make_request('POST', 'batch_predict', json={'texts': texts}) def health_check(self) -> Optional[Dict]: """健康检查""" try: response = requests.get(f"{self.base_url}/health", timeout=5) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"健康检查失败: {e}") return None

这个增强版的客户端包含了重试机制、错误处理和更完善的日志，更适合生产环境使用。

6. 监控与维护

6.1 添加访问日志

为了监控API的使用情况，我们可以添加详细的访问日志：

# 在main.py中添加日志中间件 from flask import g import logging from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('api_access.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) @app.before_request def before_request(): """记录请求开始时间""" g.start_time = time.time() @app.after_request def after_request(response): """记录请求日志""" if request.path != '/health': # 健康检查不记录详细日志 client_id = 'unknown' # 尝试从token中提取client_id if 'Authorization' in request.headers: auth_header = request.headers['Authorization'] if auth_header.startswith('Bearer '): token = auth_header.split(' ')[1] try: # 不解码，只提取client_id（避免验证开销） # 实际中可能需要更安全的方式 pass except: pass # 计算处理时间 process_time = time.time() - g.start_time log_data = { 'timestamp': datetime.now().isoformat(), 'method': request.method, 'path': request.path, 'status': response.status_code, 'process_time': round(process_time, 3), 'client_ip': request.remote_addr, 'user_agent': request.user_agent.string[:100] if request.user_agent else '', } logger.info(f"API访问: {log_data}") return response

6.2 使用Supervisor管理服务

我们使用Supervisor来管理服务进程。修改Supervisor配置文件，添加我们的新服务：

; /etc/supervisor/conf.d/structbert_jwt.conf [program:structbert_jwt_api] command=/root/miniconda3/envs/torch28/bin/python /root/nlp_structbert_sentiment-classification_chinese-base/app/main_with_jwt.py directory=/root/nlp_structbert_sentiment-classification_chinese-base autostart=true autorestart=true startretries=3 user=root redirect_stderr=true stdout_logfile=/var/log/structbert_jwt_api.log stdout_logfile_maxbytes=10MB stdout_logfile_backups=5 environment=PYTHONUNBUFFERED="1"

然后重新加载Supervisor配置：

supervisorctl reread supervisorctl update supervisorctl start structbert_jwt_api

6.3 监控API使用情况

我们可以创建一个简单的监控面板，显示API的使用统计：

# monitoring_dashboard.py from collections import defaultdict from datetime import datetime, timedelta import sqlite3 from flask import Flask, render_template_string import threading import time app = Flask(__name__) # 简单的内存存储（生产环境建议用数据库） api_stats = { 'total_requests': 0, 'successful_requests': 0, 'failed_requests': 0, 'by_endpoint': defaultdict(int), 'by_client': defaultdict(int), 'response_times': [] } def record_request(endpoint, client_id, success, response_time): """记录API请求统计""" api_stats['total_requests'] += 1 if success: api_stats['successful_requests'] += 1 else: api_stats['failed_requests'] += 1 api_stats['by_endpoint'][endpoint] += 1 api_stats['by_client'][client_id] += 1 api_stats['response_times'].append({ 'timestamp': datetime.now(), 'response_time': response_time, 'endpoint': endpoint }) # 只保留最近1000条记录 if len(api_stats['response_times']) > 1000: api_stats['response_times'] = api_stats['response_times'][-1000:] @app.route('/stats') def show_stats(): """显示API统计信息""" # 计算平均响应时间 if api_stats['response_times']: avg_time = sum(r['response_time'] for r in api_stats['response_times']) / len(api_stats['response_times']) else: avg_time = 0 # 计算成功率 if api_stats['total_requests'] > 0: success_rate = (api_stats['successful_requests'] / api_stats['total_requests']) * 100 else: success_rate = 0 # 生成HTML页面" <!DOCTYPE html> <html> <head> <title>StructBERT API监控面板</title> <style> body { font-family: Arial, sans-serif; margin: 20px; } .stats { background: #f5f5f5; padding: 20px; border-radius: 5px; margin-bottom: 20px; } .metric { display: inline-block; margin-right: 30px; } .metric-value { font-size: 24px; font-weight: bold; color: #333; } .metric-label { font-size: 14px; color: #666; } table { width: 100%; border-collapse: collapse; } th, td { padding: 10px; text-align: left; border-bottom: 1px solid #ddd; } th { background-color: #f2f2f2; } </style> </head> <body> <h1>StructBERT API监控面板</h1> <div> <div> <div>{{ total_requests }}</div> <div>总请求数</div> </div> <div> <div>{{ success_rate|round(2) }}%</div> <div>成功率</div> </div> <div> <div>{{ avg_time|round(3) }}s</div> <div>平均响应时间</div> </div> </div> <h2>按接口统计</h2> <table> <tr><th>接口</th><th>调用次数</th></tr> {% for endpoint, count in by_endpoint.items() %} <tr><td>{{ endpoint }}</td><td>{{ count }}</td></tr> {% endfor %} </table> <h2>按客户端统计</h2> <table> <tr><th>客户端</th><th>调用次数</th></tr> {% for client, count in by_client.items() %} <tr><td>{{ client }}</td><td>{{ count }}</td></tr> {% endfor %} </table> </body> </html> """ return render_template_string(html, total_requests=api_stats['total_requests'], success_rate=success_rate, avg_time=avg_time, by_endpoint=dict(api_stats['by_endpoint']), by_client=dict(api_stats['by_client']) ) def start_monitoring(): """启动监控服务""" app.run(host='0.0.0.0', port=8081) if __name__ == '__main__': # 在实际使用中，这个监控服务应该单独运行 # 这里只是示例 pass

7. 总结

通过给StructBERT中文情感分析API添加JWT认证，我们实现了一个适合企业内网使用的安全方案。让我们回顾一下关键点：

7.1 实现的核心功能

  1. 安全的身份验证：使用JWT token确保只有授权的系统可以调用API
  2. 灵活的权限控制：可以在token中携带权限信息，实现细粒度的访问控制
  3. 完善的错误处理：提供了清晰的错误信息，便于客户端处理
  4. 详细的访问日志：记录所有API调用，便于监控和审计
  5. 健壮的客户端：包含重试机制和错误处理，适合生产环境

7.2 企业部署建议

在实际企业部署时，我建议：

  1. 使用专门的认证服务：不要像示例中那样直接生成token，应该有一个统一的认证服务
  2. 定期轮换密钥：定期更换RSA密钥对，增强安全性
  3. 实现token刷新机制：使用refresh token来获取新的access token
  4. 添加速率限制：防止单个客户端过度调用API
  5. 使用API网关：在企业级部署中，建议使用API网关来统一管理认证、限流、监控等功能

7.3 扩展可能性

这个方案还可以进一步扩展：

  • 多租户支持：为不同的部门或团队提供独立的访问控制
  • 使用统计：记录每个客户端的调用次数和资源使用情况
  • 自动扩缩容：根据API调用量自动调整服务实例数量
  • 集成企业SSO：与企业的单点登录系统集成

7.4 开始使用

如果你已经部署了StructBERT情感分析服务，按照本文的步骤，大概30分钟就能完成JWT认证的添加。关键步骤是：

  1. 生成RSA密钥对
  2. 修改API服务代码，添加JWT验证
  3. 创建token生成工具
  4. 更新客户端代码，在请求中携带token
  5. 配置Supervisor管理新服务

这样改造后，你的StructBERT服务就从一个简单的演示工具，变成了一个可以安全集成到企业系统中的生产级API服务。无论是分析客户反馈、监控社交媒体情绪，还是评估客服质量，都可以放心地使用了。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 ZEEKLOG星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

Read more

可以在命令行通过大模型使用上下文协议（MCP）与外部工具交互的软件：小巧的MCPHost

可以在命令行通过大模型使用上下文协议（MCP）与外部工具交互的软件：小巧的MCPHost

小巧的MCPHost MCPHost 可以在命令行下使用，使大型语言模型（LLM）能够通过模型上下文协议（MCP）与外部工具进行交互。目前支持Claude 3.5 Sonnet和Ollama等。本次实践使用自己架设的Deepseek v3模型，跑通了Time MCP服务。  官网：GitHub - mark3labs/mcphost: A CLI host application that enables Large Language Models (LLMs) to interact with external tools through the Model Context Protocol (MCP). 下载安装 使用非常方便，直接下载解压即可使用。官网提供Windows、Linux和MacOS三个系统的压缩包： https://github.com/

By Ne0inhk
实战篇：Python开发monogod数据库mcp server看完你就会了

实战篇：Python开发monogod数据库mcp server看完你就会了

原创不易,请关注公众号：【爬虫与大模型开发】,大模型的应用开发之路，整理了大模型在现在的企业级应用的实操及大家需要注意的一些AI开发的知识点！持续输出爬虫与大模型的相关文章。 前言 目前mcp协议是给deepseek大模型插上工具链的翅膀，让大模型不仅拥有超高的推理和文本生成能力，还能具备执行大脑意识的工具能力！ 如何开发一个mcp? mcp是一种协议，指的是模型上下文协议 (Model Context Protocol)。 官方结成的mcp https://github.com/modelcontextprotocol/python-sdk mcp库 pip install mcp from mcp.server.fastmcp import FastMCP 我们先来做一个简单的案例 from mcp.server.fastmcp import FastMCP import requests mcp = FastMCP("spider") @mcp.tool() def crawl(

By Ne0inhk
AI Agent新范式：FastGPT+MCP协议实现工具增强型智能体构建

AI Agent新范式：FastGPT+MCP协议实现工具增强型智能体构建

AI Agent新范式：FastGPT+MCP协议实现工具增强型智能体构建 作者：高瑞冬 本文目录 * AI Agent新范式：FastGPT+MCP协议实现工具增强型智能体构建 * 一、MCP协议简介 * 二、创建MCP工具集 * 1. 获取MCP服务地址 * 2. 在FastGPT中创建MCP工具集 * 三、测试MCP工具 * 四、AI模型调用MCP工具 * 1. 调用单个工具 * 2. 调用整个工具集 * 五、私有化部署支持 * 1. 环境准备 * 2. 修改docker-compose.yml文件 * 3. 修改FastGPT配置 * 4. 重启服务 * 六、使用MCP-Proxy集成多个MCP服务 * 1. MCP-Proxy简介 * 2. 安装MCP-Proxy * 3. 配置MCP-Proxy * 4. 将MCP-Proxy与FastGPT集成 * 5. 高级配置

By Ne0inhk
【大模型实战篇】基于Claude MCP协议的智能体落地示例

【大模型实战篇】基于Claude MCP协议的智能体落地示例

1. 背景         之前我们在《MCP(Model Context Protocol) 大模型智能体第一个开源标准协议》一文中，介绍了MCP的概念，虽然了解了其概念、架构、解决的问题，但还缺少具体的示例，来帮助进一步理解整套MCP框架如何落地。         今天我们基于claude的官方例子--获取天气预报【1】，来理解MCP落地的整条链路。 2. MCP示例         该案例是构建一个简单的MCP天气预报服务器，并将其连接到主机，即Claude for Desktop。从基本设置开始，然后逐步发展到更复杂的使用场景。         大模型虽然能力非常强，但其弊端就是内容是过时的，这里的过时不是说内容很旧，只是表达内容具有非实时性。比如没有获取天气预报和严重天气警报的能力。因此我们将使用MCP来解决这一问题。         构建一个服务器，该服务器提供两个工具：获取警报（get-alerts）和获取预报（get-forecast）。然后，将该服务器连接到MCP主机（在本例中为Claude for Desktop）。         首先我们配置下环

By Ne0inhk