报错信息如下:
[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_URL
params = {
'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_time
if 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 = 0
for 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"达到内容长度限制,停止添加更多新闻")
break
processed_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_time
if 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_URL
params = {
'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_time
if 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 = 0
for 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"达到内容长度限制,停止添加更多新闻")
break
processed_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_time
if 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): 是否包含已完赛的比赛,默认True
include_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_URL
params = {
'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_time
if 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 = 0
finished_games = 0
upcoming_games = 0
current_length = 0
for 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:
continue
if 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 += 1
if status_info['is_finished']:
finished_games += 1
else:
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}")
break
processed_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:
break
if 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_count
home_count = sum(1 for game in team_games if game['is_home'])
away_count = len(team_games) - home_count
summary = {
"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 = None
for 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]
break
if test_team:
break
if 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()