电视节目追踪服务器

Trakt MCP服务器是一个生产就绪的AI助手服务,提供无缝访问Trakt.tv API的功能,实现智能电视节目追踪、观看列表管理和个性化观看推荐。

  • by fab-codes
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 fab-codes/trakt-mcp-server

使用说明

Trakt MCP服务器

一个生产就绪的模型上下文协议(MCP)服务器,为AI助手提供无缝访问 Track.TV API,实现智能电视节目跟踪、观看列表管理和个性化观看推荐。

特性

  • 观看历史跟踪:完整的观看历史记录,包括剧集计数、时间戳和进度跟踪
  • 智能观察列表管理:在所有设备上添加、删除和组织节目
  • 高级节目发现:利用Trakt丰富的数据库和趋势显示进行强大的搜索
  • 双重运输支持:生产就绪的HTTP(流式传输)和开发友好的stdio模式
  • 异步架构:使用FastMCP和httpx构建,用于高性能异步操作
  • 生命周期管理:使用异步上下文管理器自动初始化和清理资源
  • 全面的错误处理:身份验证、网络和API错误的详细异常层次结构
  • 结构化日志记录:用于监控和调试的可配置日志系统

建筑

项目结构

trakt-mcp-server/
├── main.py                      # Application entry point with transport selection
├── src/
│   ├── server.py               # FastMCP server with lifespan management
│   ├── api/
│   │   └── __init__.py         # TraktAPIClient with async HTTP operations
│   ├── config/
│   │   └── __init__.py         # Configuration management and validation
│   ├── exceptions/
│   │   └── __init__.py         # Custom exception hierarchy
│   ├── formatters/
│   │   └── __init__.py         # Output formatters for tool responses
│   ├── tools/
│   │   ├── __init__.py         # Tool registration system
│   │   ├── history.py          # Watch history tools
│   │   ├── watchlist.py        # Watchlist management tools
│   │   └── discovery.py        # Search and trending tools
│   └── utils/
│       └── logger.py           # Structured logging configuration
├── requirements.txt            # Python dependencies
├── .env.example               # Environment configuration template
└── README.md                  # This file

核心组件

  • FastMCP服务器:利用FastMCP进行声明性工具注册和自动类型验证
  • 异步API客户端:基于httpx的客户端,具有连接池和自动重试功能
  • 寿命背景:确保在整个服务器生命周期中正确初始化和清理资源
  • 工具类别:组织为历史、观察列表和发现模块,以提高可维护性

可用工具

历史记录工具

get_watched_shows

检索具有详细查看统计信息的完整手表历史记录。

退货:

  • 所有观看的电视节目,包括剧集数
  • 上次观看的时间戳和总播放次数
  • 查看每个系列的进度
  • 格式化JSON响应

使用案例:

  • “我最近看了什么节目?”
  • “我看过《绝命毒师》吗?”
  • 推荐过滤(避免推荐观看的内容)

监视列表工具

get_watchlist

获取用户精心策划的观察列表,这是个性化推荐的主要来源。

退货:

  • 所有带有元数据的书签节目
  • 标题、年份、评级、流派
  • 添加到监视列表时的时间戳

使用案例:

  • “今晚我该看什么?”
  • “显示我保存的节目”
  • 用户自己列表中的优先级建议

add_to_watchlist

通过跨设备同步将电视节目添加到用户的观看列表中。

参数:

  • show_id (字符串,必填):Trakt数字ID来自 search_shows
    • 图案: ^[0-9]+$ (已验证)

退货: 确认已添加演出次数

工作流程:

  1. 呼叫 search_shows 查找节目
  2. 从结果中提取Trakt ID
  3. 呼叫 add_to_watchlist 使用该ID

要求: 具有监视列表权限的OAuth访问令牌

remove_from_watchlist

从观看列表中删除节目,而不影响观看历史记录。

参数:

  • show_id (字符串,必填):Trakt数字ID

退货: 确认已删除的节目数量


发现工具

search_shows

按标题或关键字搜索Trakt的庞大数据库。

参数:

  • query (字符串,必填):搜索词(最小长度:1)

退货:

  • 前10个匹配的节目
  • Trakt ID、标题、年份、评级
  • 分数排序结果

使用案例:

  • “搜索办公室”
  • 查找监视列表操作的节目ID
  • “有一个叫《黑暗》的节目吗?”

get_trending_shows

根据社区活动获取实时趋势显示。

参数:

  • limit (整数,可选):结果数量(1-20,默认值:10)

退货:

  • 当前热门节目
  • 观众计数和评级
  • 实时社区参与数据

使用案例:

  • “现在的趋势是什么?”
  • “人们在看什么?”
  • 时代精神建议

先决条件

  • Python 3.8+ (推荐:3.11+以获得最佳异步性能)
  • Trat.tv帐户 (免费注册 Track.TV)
  • Trakt API证书 (客户端ID和OAuth访问令牌)

安装

1.克隆和设置

git clone <repository-url>
cd trakt-mcp-server

2.创建虚拟环境

python -m venv venv

# Windows
venv\Scripts\activate

# macOS/Linux
source venv/bin/activate

3.安装依赖项

pip install -r requirements.txt

依赖关系:

  • mcp -用于MCP服务器实现的FastMCP框架
  • httpx -Trakt API的高性能异步HTTP客户端
  • python-dotenv -环境变量管理
  • pydantic -数据验证(FastMCP的依赖性)

4.配置环境

cp .env.example .env

编辑 .env 使用您的凭据:

# Required: Trakt API Credentials
TRAKT_CLIENT_ID=your_trakt_client_id_here
TRAKT_ACCESS_TOKEN=your_oauth_access_token_here
TRAKT_API_VERSION=2

# Optional: Logging Configuration
LOG_LEVEL=INFO  # Options: DEBUG, INFO, WARNING, ERROR, CRITICAL

正在获取Trakt API凭据

分步指南

  1. 创建应用

    • 引导到 Trakt.tv应用程序
    • 点击“新建应用程序”
    • 填写应用程序详细信息(名称、描述、重定向URI)
  2. 获取客户端ID

    • 复制 客户端ID 从您的申请页面
    • 增添 .env 作为 TRAKT_CLIENT_ID
  3. 生成访问令牌

    • 在您的应用程序设置中,生成OAuth令牌
    • 所需权限: public, read, write (用于观察名单管理)
    • 复制生成的令牌
    • 增添 .env 作为 TRAKT_ACCESS_TOKEN
  4. 验证配置

    python -c "from src.config import AppConfig; print('✓ Configuration valid')"
    

用法

运输方式

服务器支持针对不同用例优化的两种传输协议:

标准运输(发展)

最适合本地开发和MCP客户端集成(Claude Desktop等)

python main.py

特点:

  • 每个客户都有专门的流程
  • 通过标准流输入/输出
  • 非常适合单用户本地开发
  • 未指定标志时的默认模式

流式HTTP传输(生产)

建议用于生产部署和网络访问

python main.py --http

特点:

  • 通过HTTP进行完全双向通信
  • 多个并发客户端支持
  • 服务器在上运行 0.0.0.0:8000
  • 支持流式响应
  • 生产级可扩展性

连接:

HTTP Endpoint: http://localhost:8000
Protocol: Streamable HTTP (MCP standard)

作为模块运行

# stdio mode
python -m trakt_mcp_server

# HTTP mode
python -m trakt_mcp_server --http

配置

环境变量

变量 必填 默认 描述
TRAKT_CLIENT_ID 是 - Trakt API客户端ID
TRAKT_ACCESS_TOKEN 是 - OAuth访问令牌
TRAKT_API_VERSION 是的 2 Trakt API版本
LOG_LEVEL 没有 INFO 记录冗长

API客户端配置

TraktAPIClient已在中配置 src/api/__初始化__南美国家巴拉圭的缩写(Paraguay):

httpx.AsyncClient(
    timeout=30.0,           # Request timeout
    follow_redirects=True,   # Auto-follow redirects
    headers={
        "Content-Type": "application/json",
        "User-Agent": "TraktMCPServer/1.0.0",
        "trakt-api-version": "2",
        "trakt-api-key": TRAKT_CLIENT_ID,
        "Authorization": f"Bearer {TRAKT_ACCESS_TOKEN}"
    }
)

MCP客户端集成

支持的客户

此服务器实现 模型上下文协议 并与任何兼容的客户端配合使用:

  • 自定义MCP客户端 -任何实施MCP规范的客户
  • 开发工具 -MCP检查员,测试框架

集成步骤

1.启动服务器

# For local MCP clients
python main.py

# For network-accessible deployment
python main.py --http

2.配置客户端

对于stdio传输(Claude Desktop示例):

{
  "mcpServers": {
    "trakt": {
      "command": "python",
      "args": ["main.py"],
      "cwd": "/path/to/trakt-mcp-server",
      "env": {
        "TRAKT_CLIENT_ID": "your_client_id",
        "TRAKT_ACCESS_TOKEN": "your_token"
      }
    }
  }
}

对于HTTP传输:

{
  "mcpServers": {
    "trakt": {
      "url": "http://localhost:8000",
      "transport": "streamable-http"
    }
  }
}

3.可用工具

连接后,客户端可以访问所有6个工具:

  • get_watched_shows
  • get_watchlist
  • add_to_watchlist
  • remove_from_watchlist
  • search_shows
  • get_trending_shows

日志记录

配置

通过环境变量设置日志记录级别:

LOG_LEVEL=DEBUG    # Maximum verbosity
LOG_LEVEL=INFO     # Standard operations (default)
LOG_LEVEL=WARNING  # Warnings and errors only
LOG_LEVEL=ERROR    # Errors only
LOG_LEVEL=CRITICAL # Critical failures only

日志输出

记录仪(src/utils/logger.py)提供结构化输出:

2025-01-15 14:30:22 [INFO] src.server: Initializing Trakt API client...
2025-01-15 14:30:22 [INFO] src.api: TraktAPIClient initialized successfully
2025-01-15 14:30:23 [INFO] main: Starting stdio server...

调试模式

要进行故障排除,请启用调试日志记录:

LOG_LEVEL=DEBUG python main.py

提供详细信息:

  • HTTP请求/响应详细信息
  • API端点调用
  • 工具调用参数
  • 异常堆栈跟踪

错误处理

异常层次结构

服务器实现了一个全面的错误处理系统(src/异常/__初始化__南美国家巴拉圭的缩写(Paraguay)):

TraktAPIError               # Base exception
├── AuthenticationError     # 401 - Invalid credentials
├── ResourceNotFoundError   # 404 - Show/resource not found
└── NetworkError           # Connection/timeout errors

错误响应

所有错误包括:

  1. 已登录 具有完整的上下文和堆栈跟踪
  2. 格式化的 作为用户友好的消息
  3. 归还 通过MCP协议连接到客户端

例子:

# Invalid credentials
AuthenticationError: "Invalid Trakt credentials. Check TRAKT_ACCESS_TOKEN."

# Network timeout
NetworkError: "Failed to connect to Trakt API. Check network connection."

# Show not found
ResourceNotFoundError: "Show ID 99999999 not found in Trakt database."

自动检索

httpx客户端会自动处理:

  • 连接池
  • 重定向以下内容
  • 超时管理(默认30秒)

发展

代码组织

src/
├── server.py          # FastMCP server + lifespan management
├── api/              # API client layer
├── config/           # Configuration + validation
├── exceptions/       # Custom exception types
├── formatters/       # Response formatting logic
├── tools/            # MCP tool definitions (modular)
│   ├── history.py    # Watch history tools
│   ├── watchlist.py  # Watchlist management
│   └── discovery.py  # Search + trending
└── utils/            # Logging utilities

添加新工具

使用FastMCP装饰器注册工具:

@mcp.tool()
async def my_new_tool(
    ctx: Context[ServerSession, AppContext],
    param: Annotated[str, Field(description="Parameter description")]
) -> str:
    """
    Tool description for AI assistant.

    Detailed explanation of what the tool does.
    """
    api = ctx.request_context.lifespan_context.api_client
    result = await api.my_api_method(param)
    return format_result(result)

注册 src/工具/__初始化__南美国家巴拉圭的缩写(Paraguay):

def register_all_tools(mcp: FastMCP) -> None:
    register_watchlist_tools(mcp)
    register_history_tools(mcp)
    register_discovery_tools(mcp)
    register_my_new_tools(mcp)  # Add this

贡献

欢迎投稿!拜托:

  1. 分叉存储库
  2. 创建要素分支
  3. 遵循现有的代码风格和模式
  4. 添加新功能的测试
  5. 提交拉取请求