报错信息如下:
[unsupported] Python 3.1 (mcp-calculator-main) (3) C:\Users(xsshu\AppData\Local\Programs\Python\Python313\python.exe
[unsupported] Python 3.1 C:\Users\xsshu\AppData\Local\Programs\Python\Python311\python.exe
这条输出显示了两个 Python 环境,但都被标记为 [unsupported]
,并且显示内容存在一些混乱(可能是某个 IDE 或工具的环境列表),让我们理一下目前的状况并给出明确的解决步骤。
🔍 当前情况分析
你可能有以下环境:
Python 3.13
(Python313
)—— ⚠️ 这是未来开发版,目前很多库(包括 mcp)可能不支持。Python 3.11
(Python311
)—— ✅ 支持match-case
,推荐使用这个版本。- 报错中曾使用
Python 3.9
—— ❌ 不支持match-case
,不能运行mcp
库。 [unsupported]
标签可能来自你的 IDE(例如 PyCharm)表示环境配置不完整或不兼容。
✅ 解决方案(标准步骤)
✅ 步骤 1:确认 Python 3.11
是可用的
打开命令行(Win+R 输入 cmd
),输入:
C:\Users\xsshu\AppData\Local\Programs\Python\Python311\python.exe --version
应输出:
Python 3.11.x
如果有输出说明这个 Python 可正常使用。
✅ 步骤 2:用 Python 3.11 创建虚拟环境(推荐)
你可以这样创建虚拟环境:
cd C:\Users\xsshu\Desktop\mcp-calculator-main
C:\Users\xsshu\AppData\Local\Programs\Python\Python311\python.exe -m venv venv
激活虚拟环境(Windows 命令行):
venv\Scripts\activate
激活后你会看到命令提示符变成:
(venv) C:\Users\xsshu\Desktop\mcp-calculator-main>
✅ 步骤 3:重新安装依赖
确保你在虚拟环境中,执行:
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
✅ 步骤 4:运行脚本
仍在虚拟环境中执行你的脚本:
python juhe_news_mcp.py
🔄 如果你使用的是 PyCharm 或 VS Code:
🔄 PyCharm 设置解释器步骤:
-
进入:File > Settings > Project: xxx > Python Interpreter
-
点击右上角齿轮 > Add…
-
选择 “Existing Environment”,指定路径为:
C:\Users\xsshu\Desktop\mcp-calculator-main\venv\Scripts\python.exe
-
点击 OK
✅ 总结
问题 | 原因 | 解决方式 |
---|---|---|
match-case 报错 | Python 版本太低(❤️.10) | 用 Python 3.11 运行 |
[unsupported] | IDE识别不到有效环境 | 用 Python 3.11 创建虚拟环境 |
mcp 无法导入 | pip 安装环境错乱或模块路径错 | 用正确 Python 和虚拟环境装依赖 |
实例1:聚合数据查询实时新闻
from mcp.server.fastmcp import FastMCP
import sys
import logging
import math
import random
import requests
import os
import time# ---------------------- 日志配置 ----------------------
logger = logging.getLogger('MCP_Server')
logging.basicConfig(level=logging.INFO)# ---------------------- 修复 Windows 控制台中文乱码问题 ----------------------
if sys.platform == 'win32':sys.stderr.reconfigure(encoding='utf-8')sys.stdout.reconfigure(encoding='utf-8')# ---------------------- 聚合数据新闻API配置 ----------------------
JUHE_NEWS_API_KEY = 'xxxxxxxxxxx' # 请替换为您的密钥
JUHE_NEWS_BASE_URL = 'http://v.juhe.cn/toutiao/index'# 新闻类型映射
NEWS_TYPE_MAPPING = {'top': '头条','guonei': '国内','guoji': '国际','yule': '娱乐','tiyu': '体育','junshi': '军事','keji': '科技','caijing': '财经','shishang': '时尚'
}# ---------------------- 创建 MCP Server 对象 ----------------------
mcp = FastMCP("MyTools")# 🔧 工具 1️⃣:计算器工具
@mcp.tool()
def calculator(python_expression: str) -> dict:"""数学表达式计算器功能说明:计算任意合法 Python 表达式的结果,适合用于数学推理。内置 math 和 random 模块,可以使用如 math.pi、random.randint 等。使用场景:当用户询问如"直径为 8 的圆面积是多少"、"√2+5是多少"等数学问题时,自动调用此工具。参数说明:python_expression (str): 要计算的 Python 表达式,例如 "math.pi * 4 ** 2"返回值:dict: {"success": True, "result": 计算结果}"""result = eval(python_expression)logger.info(f"计算表达式: {python_expression} = {result}")return {"success": True, "result": result}# 🔧 工具 2️⃣:中文新闻搜索工具
@mcp.tool()
def search_chinese_news(news_category: str = "top", max_articles: int = 3, page_number: int = 1, include_content: bool = False) -> dict:"""中文新闻搜索工具功能说明:获取指定分类的中文新闻,支持多种新闻分类和分页,数据来源于聚合数据API。支持获取新闻标题、内容、作者、日期等完整信息。使用场景:当用户询问"查看今天的头条新闻"、"获取科技新闻"、"查看国内新闻"等需要中文新闻时,自动调用此工具。参数说明:news_category (str): 新闻分类,可选值: top(头条), guonei(国内), guoji(国际), yule(娱乐), tiyu(体育), junshi(军事), keji(科技), caijing(财经), shishang(时尚)max_articles (int): 最大返回文章数量,默认为3篇,最大不超过5篇page_number (int): 页码,默认第1页include_content (bool): 是否包含新闻内容,默认False(仅标题),True时返回完整内容返回值:dict: {"success": True, "news": [{"title": "标题", "content": "内容", "author": "作者", "date": "日期", "category": "分类", "url": "链接", "uniquekey": "唯一标识"}]}"""# 检查API密钥if JUHE_NEWS_API_KEY == 'YOUR_API_KEY_HERE' or not JUHE_NEWS_API_KEY:return {"success": False, "error": "请先配置聚合数据API密钥"}# 验证分类参数if news_category not in NEWS_TYPE_MAPPING:news_category = "top"# 限制文章数量和页码max_articles = min(max_articles, 5)page_number = max(1, page_number)try:# 构造API请求 - 根据官方文档完善参数url = JUHE_NEWS_BASE_URLparams = {'type': news_category,'key': JUHE_NEWS_API_KEY,'page': page_number,'page_size': max_articles,'is_filter': 1 # 过滤垃圾信息}logger.info(f"请求新闻API: {url}, 参数: {params}")# 发送请求start_time = time.time()response = requests.get(url, params=params, timeout=15)response_time = time.time() - start_timeif response.status_code != 200:logger.error(f"聚合数据API错误: {response.status_code}")return {"success": False, "error": f"API请求失败,状态码: {response.status_code}"}data = response.json()logger.info(f"API返回状态: error_code={data.get('error_code')}, reason={data.get('reason')}")if data.get('error_code') != 0:error_msg = data.get('reason', '未知错误')logger.error(f"聚合数据API返回错误: {error_msg}")return {"success": False, "error": f"API错误: {error_msg}"}result = data.get('result', {})articles = result.get('data', [])if not articles:category_name = NEWS_TYPE_MAPPING.get(news_category, news_category)return {"success": True, "news": [], "message": f"未找到 {category_name} 新闻"}# 调试:记录API返回的数据结构if articles:sample_article = articles[0]available_fields = list(sample_article.keys())logger.info(f"API返回数据字段: {available_fields}")# 处理文章数据processed_news = []total_length = 0for i, article in enumerate(articles):# 获取基本信息title = article.get('title', '无标题')[:120]author = article.get('author_name', '未知作者')date = article.get('date', '未知日期')category_name = NEWS_TYPE_MAPPING.get(news_category, news_category)url = article.get('url', '')uniquekey = article.get('uniquekey', '')thumbnail = article.get('thumbnail_pic_s', '')# 获取新闻内容content = ""if include_content:# 根据官方文档,content字段包含新闻内容raw_content = article.get('content', '')if raw_content:content = str(raw_content)[:300] # 限制内容长度else:content = "暂无详细内容"else:content = "如需查看详细内容,请设置include_content=True"news_data = {"title": title,"content": content,"author": author,"date": date,"category": category_name,"url": url,"uniquekey": uniquekey,"thumbnail": thumbnail}# 检查总长度是否超过MCP限制news_str = str(news_data)if total_length + len(news_str) > 1200:logger.info(f"达到内容长度限制,停止添加更多新闻")breakprocessed_news.append(news_data)total_length += len(news_str)logger.info(f"中文新闻: {category_name}, 页码: {page_number}, 返回 {len(processed_news)} 篇, 响应时间: {response_time:.2f}秒")return {"success": True,"news": processed_news,"category": category_name,"page": page_number,"total_found": len(articles),"include_content": include_content,"api_response_time": f"{response_time:.2f}秒"}except requests.exceptions.Timeout:logger.error(f"聚合数据API请求超时: {news_category}")return {"success": False, "error": "请求超时,请稍后重试"}except requests.exceptions.RequestException as e:logger.error(f"聚合数据API网络错误: {str(e)}")return {"success": False, "error": "网络连接错误"}except Exception as e:logger.error(f"获取中文新闻出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🔧 工具 3️⃣:新闻详情查询工具(基于uniquekey)
@mcp.tool()
def get_news_detail_by_uniquekey(news_uniquekey: str) -> dict:"""根据新闻唯一标识获取详细内容功能说明:通过新闻的uniquekey获取该新闻的完整详细信息,包括标题、内容、作者等。这是获取特定新闻详细内容的推荐方式。使用场景:当用户想要查看某条特定新闻的详细内容时,可以使用此工具。通常配合新闻搜索工具使用,先搜索新闻获取uniquekey,再查询详情。参数说明:news_uniquekey (str): 新闻的唯一标识符,从新闻搜索结果中获取返回值:dict: {"success": True, "title": "标题", "content": "详细内容", "author": "作者", "date": "日期", "url": "链接"}"""# 检查API密钥if JUHE_NEWS_API_KEY == 'YOUR_API_KEY_HERE' or not JUHE_NEWS_API_KEY:return {"success": False, "error": "请先配置聚合数据API密钥"}# 验证参数if not news_uniquekey or not isinstance(news_uniquekey, str):return {"success": False, "error": "请提供有效的新闻uniquekey"}try:# 构造API请求 - 使用uniquekey查询特定新闻详情url = "http://v.juhe.cn/toutiao/content" # 新闻详情查询接口params = {'key': JUHE_NEWS_API_KEY,'uniquekey': news_uniquekey}logger.info(f"查询新闻详情: uniquekey={news_uniquekey}")# 发送请求start_time = time.time()response = requests.get(url, params=params, timeout=15)response_time = time.time() - start_timeif response.status_code != 200:logger.error(f"新闻详情API错误: {response.status_code}")return {"success": False, "error": f"API请求失败,状态码: {response.status_code}"}data = response.json()if data.get('error_code') != 0:error_msg = data.get('reason', '未知错误')logger.error(f"新闻详情API返回错误: {error_msg}")return {"success": False, "error": f"API错误: {error_msg}"}result = data.get('result', {})if not result:return {"success": False, "error": "未找到该新闻详情"}# 提取详细信息title = result.get('title', '无标题')content = result.get('content', '暂无内容')detail = result.get('detail', '暂无详细信息')author = result.get('author_name', '未知作者')date = result.get('date', '未知日期')category = result.get('category', '未知分类')url = result.get('url', '')# 由于MCP返回值长度限制,适当截取内容if len(content) > 800:content = content[:800] + "...[内容过长已截取]"logger.info(f"成功获取新闻详情: {title[:30]}..., 响应时间: {response_time:.2f}秒")return {"success": True,"uniquekey": news_uniquekey,"title": title,"content": content,"detail": detail,"author": author,"date": date,"category": category,"url": url,"api_response_time": f"{response_time:.2f}秒"}except requests.exceptions.Timeout:logger.error(f"新闻详情API请求超时: {news_uniquekey}")return {"success": False, "error": "请求超时,请稍后重试"}except requests.exceptions.RequestException as e:logger.error(f"新闻详情API网络错误: {str(e)}")return {"success": False, "error": "网络连接错误"}except Exception as e:logger.error(f"获取新闻详情出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🔧 工具 4️⃣:新闻URL内容提取工具(保留作为备用)
@mcp.tool()
def extract_news_content_from_url(news_url: str) -> dict:"""从新闻链接提取内容(备用方案)功能说明:通过新闻URL直接抓取网页内容,作为获取新闻详情的备用方案。注意:由于网站反爬虫机制,成功率可能较低,推荐优先使用uniquekey查询方式。使用场景:当无法通过uniquekey获取详情时的备用方案。参数说明:news_url (str): 新闻的完整URL链接返回值:dict: {"success": True, "content": "提取的内容", "url": "原链接"}"""if not news_url or not news_url.startswith('http'):return {"success": False, "error": "无效的新闻链接"}try:# 设置请求头,模拟浏览器访问headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36','Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8','Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3','Connection': 'keep-alive',}response = requests.get(news_url, headers=headers, timeout=10)if response.status_code != 200:return {"success": False, "error": f"无法访问新闻链接,状态码: {response.status_code}"}# 简单的内容提取content = response.text# 由于MCP返回值长度限制,只返回前600字符if len(content) > 600:content = content[:600] + "...[内容已截取]"logger.info(f"成功提取URL内容: {news_url[:50]}...")return {"success": True,"content": content,"url": news_url,"method": "URL抓取","note": "此方法为备用方案,推荐使用uniquekey查询获取更准确的内容"}except requests.exceptions.Timeout:return {"success": False, "error": "请求超时"}except requests.exceptions.RequestException as e:return {"success": False, "error": f"网络错误: {str(e)}"}except Exception as e:logger.error(f"URL内容提取出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🚀 启动 MCP Server 主程序
if __name__ == "__main__":mcp.run(transport="stdio")# 🧪 测试函数(开发调试用)
def test_news_api():"""测试新闻API功能的验证函数仅在开发阶段使用,用于验证API改进效果"""print("=" * 50)print("📰 新闻API功能测试")print("=" * 50)# 测试1:基础新闻搜索(不包含内容)print("\n🔸 测试1:基础新闻搜索")result1 = search_chinese_news(news_category="keji", max_articles=2, include_content=False)print(f"成功: {result1.get('success')}")if result1.get('success'):print(f"新闻数量: {len(result1.get('news', []))}")if result1.get('news'):first_news = result1['news'][0]print(f"标题示例: {first_news.get('title', '')[:50]}...")print(f"uniquekey: {first_news.get('uniquekey', 'N/A')}")# 测试2:包含内容的新闻搜索print("\n🔸 测试2:包含内容的新闻搜索")result2 = search_chinese_news(news_category="keji", max_articles=1, include_content=True)print(f"成功: {result2.get('success')}")if result2.get('success') and result2.get('news'):news_with_content = result2['news'][0]content = news_with_content.get('content', '')print(f"内容长度: {len(content)} 字符")print(f"内容预览: {content[:100]}..." if len(content) > 100 else f"完整内容: {content}")# 测试3:基于uniquekey的详情查询if result1.get('success') and result1.get('news') and result1['news'][0].get('uniquekey'):print("\n🔸 测试3:uniquekey详情查询")uniquekey = result1['news'][0]['uniquekey']result3 = get_news_detail_by_uniquekey(uniquekey)print(f"成功: {result3.get('success')}")if result3.get('success'):print(f"详情内容长度: {len(result3.get('content', ''))} 字符")print("\n" + "=" * 50)print("✅ 测试完成")print("=" * 50)# 如果需要测试,取消下面的注释
# test_news_api()
实例2:NBA赛事查询
from mcp.server.fastmcp import FastMCP
import sys
import logging
import math
import random
import requests
import os
import time# ---------------------- 日志配置 ----------------------
logger = logging.getLogger('MCP_Server')
logging.basicConfig(level=logging.INFO)# ---------------------- 修复 Windows 控制台中文乱码问题 ----------------------
if sys.platform == 'win32':sys.stderr.reconfigure(encoding='utf-8')sys.stdout.reconfigure(encoding='utf-8')# ---------------------- 聚合数据新闻API配置 ----------------------
JUHE_NEWS_API_KEY = '6688b0200d8fd96381fb0d6ae4e03cc5' # 请替换为您的密钥
JUHE_NEWS_BASE_URL = 'http://v.juhe.cn/toutiao/index'# 新闻类型映射
NEWS_TYPE_MAPPING = {'top': '头条','guonei': '国内','guoji': '国际','yule': '娱乐','tiyu': '体育','junshi': '军事','keji': '科技','caijing': '财经','shishang': '时尚'
}# ---------------------- 聚合数据NBA API配置 ----------------------
JUHE_NBA_API_KEY = 'xxxxxxxxxxxxxxxxxxxxxx' # NBA赛事API密钥
JUHE_NBA_BASE_URL = 'http://apis.juhe.cn/fapig/nba/query'# NBA比赛状态映射
NBA_STATUS_MAPPING = {'1': {'text': '未开赛', 'is_finished': False},'2': {'text': '进行中', 'is_finished': False}, '3': {'text': '完赛', 'is_finished': True}
}# ---------------------- 创建 MCP Server 对象 ----------------------
mcp = FastMCP("MyTools")# 🔧 工具 1️⃣:计算器工具
@mcp.tool()
def calculator(python_expression: str) -> dict:"""数学表达式计算器功能说明:计算任意合法 Python 表达式的结果,适合用于数学推理。内置 math 和 random 模块,可以使用如 math.pi、random.randint 等。使用场景:当用户询问如"直径为 8 的圆面积是多少"、"√2+5是多少"等数学问题时,自动调用此工具。参数说明:python_expression (str): 要计算的 Python 表达式,例如 "math.pi * 4 ** 2"返回值:dict: {"success": True, "result": 计算结果}"""result = eval(python_expression)logger.info(f"计算表达式: {python_expression} = {result}")return {"success": True, "result": result}# 🔧 工具 2️⃣:中文新闻搜索工具
@mcp.tool()
def search_chinese_news(news_category: str = "top", max_articles: int = 3, page_number: int = 1, include_content: bool = False) -> dict:"""中文新闻搜索工具功能说明:获取指定分类的中文新闻,支持多种新闻分类和分页,数据来源于聚合数据API。支持获取新闻标题、内容、作者、日期等完整信息。使用场景:当用户询问"查看今天的头条新闻"、"获取科技新闻"、"查看国内新闻"等需要中文新闻时,自动调用此工具。参数说明:news_category (str): 新闻分类,可选值: top(头条), guonei(国内), guoji(国际), yule(娱乐), tiyu(体育), junshi(军事), keji(科技), caijing(财经), shishang(时尚)max_articles (int): 最大返回文章数量,默认为3篇,最大不超过5篇page_number (int): 页码,默认第1页include_content (bool): 是否包含新闻内容,默认False(仅标题),True时返回完整内容返回值:dict: {"success": True, "news": [{"title": "标题", "content": "内容", "author": "作者", "date": "日期", "category": "分类", "url": "链接", "uniquekey": "唯一标识"}]}"""# 检查API密钥if JUHE_NEWS_API_KEY == 'YOUR_API_KEY_HERE' or not JUHE_NEWS_API_KEY:return {"success": False, "error": "请先配置聚合数据API密钥"}# 验证分类参数if news_category not in NEWS_TYPE_MAPPING:news_category = "top"# 限制文章数量和页码max_articles = min(max_articles, 5)page_number = max(1, page_number)try:# 构造API请求 - 根据官方文档完善参数url = JUHE_NEWS_BASE_URLparams = {'type': news_category,'key': JUHE_NEWS_API_KEY,'page': page_number,'page_size': max_articles,'is_filter': 1 # 过滤垃圾信息}logger.info(f"请求新闻API: {url}, 参数: {params}")# 发送请求start_time = time.time()response = requests.get(url, params=params, timeout=15)response_time = time.time() - start_timeif response.status_code != 200:logger.error(f"聚合数据API错误: {response.status_code}")return {"success": False, "error": f"API请求失败,状态码: {response.status_code}"}data = response.json()logger.info(f"API返回状态: error_code={data.get('error_code')}, reason={data.get('reason')}")if data.get('error_code') != 0:error_msg = data.get('reason', '未知错误')logger.error(f"聚合数据API返回错误: {error_msg}")return {"success": False, "error": f"API错误: {error_msg}"}result = data.get('result', {})articles = result.get('data', [])if not articles:category_name = NEWS_TYPE_MAPPING.get(news_category, news_category)return {"success": True, "news": [], "message": f"未找到 {category_name} 新闻"}# 调试:记录API返回的数据结构if articles:sample_article = articles[0]available_fields = list(sample_article.keys())logger.info(f"API返回数据字段: {available_fields}")# 处理文章数据processed_news = []total_length = 0for i, article in enumerate(articles):# 获取基本信息title = article.get('title', '无标题')[:120]author = article.get('author_name', '未知作者')date = article.get('date', '未知日期')category_name = NEWS_TYPE_MAPPING.get(news_category, news_category)url = article.get('url', '')uniquekey = article.get('uniquekey', '')thumbnail = article.get('thumbnail_pic_s', '')# 获取新闻内容content = ""if include_content:# 根据官方文档,content字段包含新闻内容raw_content = article.get('content', '')if raw_content:content = str(raw_content)[:300] # 限制内容长度else:content = "暂无详细内容"else:content = "如需查看详细内容,请设置include_content=True"news_data = {"title": title,"content": content,"author": author,"date": date,"category": category_name,"url": url,"uniquekey": uniquekey,"thumbnail": thumbnail}# 检查总长度是否超过MCP限制news_str = str(news_data)if total_length + len(news_str) > 1200:logger.info(f"达到内容长度限制,停止添加更多新闻")breakprocessed_news.append(news_data)total_length += len(news_str)logger.info(f"中文新闻: {category_name}, 页码: {page_number}, 返回 {len(processed_news)} 篇, 响应时间: {response_time:.2f}秒")return {"success": True,"news": processed_news,"category": category_name,"page": page_number,"total_found": len(articles),"include_content": include_content,"api_response_time": f"{response_time:.2f}秒"}except requests.exceptions.Timeout:logger.error(f"聚合数据API请求超时: {news_category}")return {"success": False, "error": "请求超时,请稍后重试"}except requests.exceptions.RequestException as e:logger.error(f"聚合数据API网络错误: {str(e)}")return {"success": False, "error": "网络连接错误"}except Exception as e:logger.error(f"获取中文新闻出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🔧 工具 3️⃣:新闻详情查询工具(基于uniquekey)
@mcp.tool()
def get_news_detail_by_uniquekey(news_uniquekey: str) -> dict:"""根据新闻唯一标识获取详细内容功能说明:通过新闻的uniquekey获取该新闻的完整详细信息,包括标题、内容、作者等。这是获取特定新闻详细内容的推荐方式。使用场景:当用户想要查看某条特定新闻的详细内容时,可以使用此工具。通常配合新闻搜索工具使用,先搜索新闻获取uniquekey,再查询详情。参数说明:news_uniquekey (str): 新闻的唯一标识符,从新闻搜索结果中获取返回值:dict: {"success": True, "title": "标题", "content": "详细内容", "author": "作者", "date": "日期", "url": "链接"}"""# 检查API密钥if JUHE_NEWS_API_KEY == 'YOUR_API_KEY_HERE' or not JUHE_NEWS_API_KEY:return {"success": False, "error": "请先配置聚合数据API密钥"}# 验证参数if not news_uniquekey or not isinstance(news_uniquekey, str):return {"success": False, "error": "请提供有效的新闻uniquekey"}try:# 构造API请求 - 使用uniquekey查询特定新闻详情url = "http://v.juhe.cn/toutiao/content" # 新闻详情查询接口params = {'key': JUHE_NEWS_API_KEY,'uniquekey': news_uniquekey}logger.info(f"查询新闻详情: uniquekey={news_uniquekey}")# 发送请求start_time = time.time()response = requests.get(url, params=params, timeout=15)response_time = time.time() - start_timeif response.status_code != 200:logger.error(f"新闻详情API错误: {response.status_code}")return {"success": False, "error": f"API请求失败,状态码: {response.status_code}"}data = response.json()if data.get('error_code') != 0:error_msg = data.get('reason', '未知错误')logger.error(f"新闻详情API返回错误: {error_msg}")return {"success": False, "error": f"API错误: {error_msg}"}result = data.get('result', {})if not result:return {"success": False, "error": "未找到该新闻详情"}# 提取详细信息title = result.get('title', '无标题')content = result.get('content', '暂无内容')detail = result.get('detail', '暂无详细信息')author = result.get('author_name', '未知作者')date = result.get('date', '未知日期')category = result.get('category', '未知分类')url = result.get('url', '')# 由于MCP返回值长度限制,适当截取内容if len(content) > 800:content = content[:800] + "...[内容过长已截取]"logger.info(f"成功获取新闻详情: {title[:30]}..., 响应时间: {response_time:.2f}秒")return {"success": True,"uniquekey": news_uniquekey,"title": title,"content": content,"detail": detail,"author": author,"date": date,"category": category,"url": url,"api_response_time": f"{response_time:.2f}秒"}except requests.exceptions.Timeout:logger.error(f"新闻详情API请求超时: {news_uniquekey}")return {"success": False, "error": "请求超时,请稍后重试"}except requests.exceptions.RequestException as e:logger.error(f"新闻详情API网络错误: {str(e)}")return {"success": False, "error": "网络连接错误"}except Exception as e:logger.error(f"获取新闻详情出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🔧 工具 4️⃣:新闻URL内容提取工具(保留作为备用)
@mcp.tool()
def extract_news_content_from_url(news_url: str) -> dict:"""从新闻链接提取内容(备用方案)功能说明:通过新闻URL直接抓取网页内容,作为获取新闻详情的备用方案。注意:由于网站反爬虫机制,成功率可能较低,推荐优先使用uniquekey查询方式。使用场景:当无法通过uniquekey获取详情时的备用方案。参数说明:news_url (str): 新闻的完整URL链接返回值:dict: {"success": True, "content": "提取的内容", "url": "原链接"}"""if not news_url or not news_url.startswith('http'):return {"success": False, "error": "无效的新闻链接"}try:# 设置请求头,模拟浏览器访问headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36','Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8','Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3','Connection': 'keep-alive',}response = requests.get(news_url, headers=headers, timeout=10)if response.status_code != 200:return {"success": False, "error": f"无法访问新闻链接,状态码: {response.status_code}"}# 简单的内容提取content = response.text# 由于MCP返回值长度限制,只返回前600字符if len(content) > 600:content = content[:600] + "...[内容已截取]"logger.info(f"成功提取URL内容: {news_url[:50]}...")return {"success": True,"content": content,"url": news_url,"method": "URL抓取","note": "此方法为备用方案,推荐使用uniquekey查询获取更准确的内容"}except requests.exceptions.Timeout:return {"success": False, "error": "请求超时"}except requests.exceptions.RequestException as e:return {"success": False, "error": f"网络错误: {str(e)}"}except Exception as e:logger.error(f"URL内容提取出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🔧 工具 5️⃣:NBA赛程赛果查询工具
@mcp.tool()
def query_nba_schedule_results(max_days: int = 7, include_finished_games: bool = True, include_upcoming_games: bool = True) -> dict:"""NBA赛程赛果查询工具功能说明:查询NBA近期的赛程安排和比赛结果,支持灵活的过滤条件。数据来源于聚合数据NBA API,包含详细的比赛信息。使用场景:当用户询问"查看NBA最近的比赛结果"、"今天有什么NBA比赛"、"NBA赛程查询"、"湖人队最近比赛"等NBA相关问题时,自动调用此工具。参数说明:max_days (int): 返回最近几天的赛程,默认7天,建议不超过10天include_finished_games (bool): 是否包含已完赛的比赛,默认Trueinclude_upcoming_games (bool): 是否包含未开赛的比赛,默认True返回值:dict: {"success": True, "league_info": {"title": "联赛名称", "season": "赛季"}, "schedule": [赛程数据], "summary": {统计摘要}}"""# 检查API密钥if not JUHE_NBA_API_KEY:return {"success": False, "error": "请先配置NBA API密钥"}# 验证参数max_days = max(1, min(max_days, 10)) # 限制在1-10天之间if not include_finished_games and not include_upcoming_games:return {"success": False, "error": "至少需要包含一种比赛类型(已完赛或未开赛)"}try:# 构造API请求url = JUHE_NBA_BASE_URLparams = {'key': JUHE_NBA_API_KEY}logger.info(f"请求NBA赛程API: {url}")# 发送请求start_time = time.time()response = requests.get(url, params=params, timeout=15)response_time = time.time() - start_timeif response.status_code != 200:logger.error(f"NBA API错误: {response.status_code}")return {"success": False, "error": f"API请求失败,状态码: {response.status_code}"}data = response.json()logger.info(f"NBA API返回状态: error_code={data.get('error_code')}, reason={data.get('reason')}")if data.get('error_code') != 0:error_msg = data.get('reason', '未知错误')logger.error(f"NBA API返回错误: {error_msg}")return {"success": False, "error": f"API错误: {error_msg}"}result = data.get('result', {})if not result:return {"success": False, "error": "未获取到NBA赛程数据"}# 提取联赛信息league_info = {"title": result.get('title', 'NBA'),"season": result.get('duration', '未知赛季')}matches = result.get('matchs', [])if not matches:return {"success": True, "league_info": league_info, "schedule": [], "message": "暂无赛程数据"}# 处理赛程数据processed_schedule = []total_games = 0finished_games = 0upcoming_games = 0current_length = 0for day_data in matches[:max_days]:date = day_data.get('date', '未知日期')week = day_data.get('week', '未知')day_games = day_data.get('list', [])if not day_games:continue# 过滤比赛filtered_games = []for game in day_games:status = game.get('status', '1')status_info = NBA_STATUS_MAPPING.get(status, {'text': '未知状态', 'is_finished': False})# 根据参数过滤比赛if status_info['is_finished'] and not include_finished_games:continueif not status_info['is_finished'] and not include_upcoming_games:continue# 处理比赛数据game_data = {"time": game.get('time_start', '未知时间'),"status": status_info['text'],"team1": game.get('team1', '未知球队1'),"team2": game.get('team2', '未知球队2'),"score1": game.get('team1_score', '-'),"score2": game.get('team2_score', '-'),"is_finished": status_info['is_finished']}filtered_games.append(game_data)total_games += 1if status_info['is_finished']:finished_games += 1else:upcoming_games += 1# 如果当天有比赛,添加到结果中if filtered_games:day_schedule = {"date": date,"week": week,"games": filtered_games}# 检查长度限制day_str = str(day_schedule)if current_length + len(day_str) > 1500: # MCP长度限制logger.info(f"达到内容长度限制,截止到{date}")breakprocessed_schedule.append(day_schedule)current_length += len(day_str)# 构建统计摘要summary = {"total_days": len(processed_schedule),"total_games": total_games,"finished_games": finished_games,"upcoming_games": upcoming_games}logger.info(f"NBA赛程查询完成: {summary['total_days']}天, {summary['total_games']}场比赛, 响应时间: {response_time:.2f}秒")return {"success": True,"league_info": league_info,"schedule": processed_schedule,"summary": summary,"filters": {"max_days": max_days,"include_finished": include_finished_games,"include_upcoming": include_upcoming_games},"api_response_time": f"{response_time:.2f}秒"}except requests.exceptions.Timeout:logger.error(f"NBA API请求超时")return {"success": False, "error": "请求超时,请稍后重试"}except requests.exceptions.RequestException as e:logger.error(f"NBA API网络错误: {str(e)}")return {"success": False, "error": "网络连接错误"}except Exception as e:logger.error(f"获取NBA赛程出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🔧 工具 6️⃣:NBA特定球队赛程查询工具
@mcp.tool()
def query_nba_team_schedule(team_name: str, max_games: int = 5) -> dict:"""NBA特定球队赛程查询工具功能说明:查询指定NBA球队的近期赛程和比赛结果。通过球队名称关键词匹配,支持中文球队名称。使用场景:当用户询问"湖人队最近的比赛"、"勇士队赛程"、"查看热火队比赛结果"等针对特定球队的NBA问题时,自动调用此工具。参数说明:team_name (str): 球队名称关键词,支持中文名称,如"湖人"、"勇士"、"热火"等max_games (int): 返回最大比赛数量,默认5场返回值:dict: {"success": True, "team_name": "球队名称", "games": [比赛数据], "summary": {统计信息}}"""# 检查API密钥if not JUHE_NBA_API_KEY:return {"success": False, "error": "请先配置NBA API密钥"}# 验证参数if not team_name or not isinstance(team_name, str):return {"success": False, "error": "请提供有效的球队名称"}team_name = team_name.strip()max_games = max(1, min(max_games, 10)) # 限制在1-10场之间try:# 先获取所有赛程数据schedule_result = query_nba_schedule_results(max_days=10, include_finished_games=True, include_upcoming_games=True)if not schedule_result.get('success'):return schedule_result # 直接返回错误信息schedule_data = schedule_result.get('schedule', [])if not schedule_data:return {"success": True, "team_name": team_name, "games": [], "message": "暂无赛程数据"}# 搜索包含指定球队的比赛team_games = []for day_data in schedule_data:date = day_data.get('date', '')week = day_data.get('week', '')games = day_data.get('games', [])for game in games:team1 = game.get('team1', '')team2 = game.get('team2', '')# 检查球队名称是否匹配(支持模糊匹配)if team_name in team1 or team_name in team2:game_info = {"date": date,"week": week,"time": game.get('time', ''),"status": game.get('status', ''),"team1": team1,"team2": team2,"score1": game.get('score1', '-'),"score2": game.get('score2', '-'),"is_finished": game.get('is_finished', False),"is_home": team_name in team1 # 判断是否主场}team_games.append(game_info)# 限制返回数量if len(team_games) >= max_games:breakif len(team_games) >= max_games:break# 统计信息finished_count = sum(1 for game in team_games if game['is_finished'])upcoming_count = len(team_games) - finished_counthome_count = sum(1 for game in team_games if game['is_home'])away_count = len(team_games) - home_countsummary = {"total_games": len(team_games),"finished_games": finished_count,"upcoming_games": upcoming_count,"home_games": home_count,"away_games": away_count}logger.info(f"球队赛程查询完成: {team_name}, 找到{len(team_games)}场比赛")if not team_games:return {"success": True, "team_name": team_name, "games": [], "message": f"未找到包含'{team_name}'的比赛"}return {"success": True,"team_name": team_name,"games": team_games,"summary": summary,"note": f"根据关键词'{team_name}'匹配到的比赛结果"}except Exception as e:logger.error(f"查询球队赛程出错: {str(e)}")return {"success": False, "error": f"处理请求时发生错误: {str(e)}"}# 🚀 启动 MCP Server 主程序
if __name__ == "__main__":mcp.run(transport="stdio")# 🧪 测试函数(开发调试用)
def test_news_api():"""测试新闻API功能的验证函数仅在开发阶段使用,用于验证API改进效果"""print("=" * 50)print("📰 新闻API功能测试")print("=" * 50)# 测试1:基础新闻搜索(不包含内容)print("\n🔸 测试1:基础新闻搜索")result1 = search_chinese_news(news_category="keji", max_articles=2, include_content=False)print(f"成功: {result1.get('success')}")if result1.get('success'):print(f"新闻数量: {len(result1.get('news', []))}")if result1.get('news'):first_news = result1['news'][0]print(f"标题示例: {first_news.get('title', '')[:50]}...")print(f"uniquekey: {first_news.get('uniquekey', 'N/A')}")# 测试2:包含内容的新闻搜索print("\n🔸 测试2:包含内容的新闻搜索")result2 = search_chinese_news(news_category="keji", max_articles=1, include_content=True)print(f"成功: {result2.get('success')}")if result2.get('success') and result2.get('news'):news_with_content = result2['news'][0]content = news_with_content.get('content', '')print(f"内容长度: {len(content)} 字符")print(f"内容预览: {content[:100]}..." if len(content) > 100 else f"完整内容: {content}")# 测试3:基于uniquekey的详情查询if result1.get('success') and result1.get('news') and result1['news'][0].get('uniquekey'):print("\n🔸 测试3:uniquekey详情查询")uniquekey = result1['news'][0]['uniquekey']result3 = get_news_detail_by_uniquekey(uniquekey)print(f"成功: {result3.get('success')}")if result3.get('success'):print(f"详情内容长度: {len(result3.get('content', ''))} 字符")print("\n" + "=" * 50)print("✅ 新闻API测试完成")print("=" * 50)# 🧪 NBA API测试函数(开发调试用)
def test_nba_api():"""测试NBA API功能的验证函数仅在开发阶段使用,用于验证NBA API功能"""print("=" * 50)print("🏀 NBA API功能测试")print("=" * 50)# 测试1:基础NBA赛程查询print("\n🔸 测试1:基础NBA赛程查询")result1 = query_nba_schedule_results(max_days=3, include_finished_games=True, include_upcoming_games=True)print(f"成功: {result1.get('success')}")if result1.get('success'):league_info = result1.get('league_info', {})print(f"联赛: {league_info.get('title', 'N/A')}")print(f"赛季: {league_info.get('season', 'N/A')}")summary = result1.get('summary', {})print(f"总天数: {summary.get('total_days', 0)}")print(f"总比赛: {summary.get('total_games', 0)}")print(f"已完赛: {summary.get('finished_games', 0)}")print(f"未开赛: {summary.get('upcoming_games', 0)}")# 显示第一天的比赛示例schedule = result1.get('schedule', [])if schedule:first_day = schedule[0]print(f"示例日期: {first_day.get('date', '')} {first_day.get('week', '')}")games = first_day.get('games', [])if games:print(f"该日比赛数: {len(games)}")first_game = games[0]print(f"比赛示例: {first_game.get('team1', '')} vs {first_game.get('team2', '')} ({first_game.get('status', '')})")# 测试2:只查询已完赛的比赛print("\n🔸 测试2:只查询已完赛比赛")result2 = query_nba_schedule_results(max_days=5, include_finished_games=True, include_upcoming_games=False)print(f"成功: {result2.get('success')}")if result2.get('success'):summary2 = result2.get('summary', {})print(f"仅已完赛比赛: {summary2.get('finished_games', 0)}场")# 测试3:球队特定查询print("\n🔸 测试3:球队特定查询")# 使用从前面测试中获取的球队名称if result1.get('success') and result1.get('schedule'):schedule = result1['schedule']test_team = Nonefor day_data in schedule:games = day_data.get('games', [])if games:first_game = games[0]team1 = first_game.get('team1', '')if team1:# 提取球队名称的关键词if '湖人' in team1:test_team = '湖人'elif '勇士' in team1:test_team = '勇士'elif '热火' in team1:test_team = '热火'else:# 取球队名称的前2个字符作为关键词test_team = team1[:2]breakif test_team:breakif test_team:result3 = query_nba_team_schedule(team_name=test_team, max_games=3)print(f"测试球队: {test_team}")print(f"成功: {result3.get('success')}")if result3.get('success'):summary3 = result3.get('summary', {})print(f"找到比赛: {summary3.get('total_games', 0)}场")print(f"主场比赛: {summary3.get('home_games', 0)}场")print(f"客场比赛: {summary3.get('away_games', 0)}场")print("\n" + "=" * 50)print("✅ NBA API测试完成")print("=" * 50)# 如果需要测试,取消下面的注释
# test_news_api()
# test_nba_api()