【小技巧】Python+PyCharm IDE 配置解释器出错,环境配置不完整或不兼容。(小智AI、MCP、聚合数据、实时新闻查询、NBA赛事查询)

发布于:2025-07-03 ⋅ 阅读:(23) ⋅ 点赞:(0)

报错信息如下:

[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 或工具的环境列表),让我们理一下目前的状况并给出明确的解决步骤


🔍 当前情况分析

你可能有以下环境:

  1. Python 3.13Python313)—— ⚠️ 这是未来开发版,目前很多库(包括 mcp)可能不支持
  2. Python 3.11Python311)—— ✅ 支持 match-case推荐使用这个版本
  3. 报错中曾使用 Python 3.9 —— ❌ 不支持 match-case不能运行 mcp
  4. [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 设置解释器步骤:

  1. 进入:File > Settings > Project: xxx > Python Interpreter

  2. 点击右上角齿轮 > Add…

  3. 选择 “Existing Environment”,指定路径为:

    C:\Users\xsshu\Desktop\mcp-calculator-main\venv\Scripts\python.exe
    
  4. 点击 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() 

网站公告

今日签到

点亮在社区的每一天
去签到