使用说明

MCP网关

基于Python的模型上下文协议（MCP）网关为YouTube自动化、谷歌趋势分析、内容安全和货币化优化提供了19个生产就绪工具。

部分 yt工厂 生态系统——充当编排器和外部API（YouTube、谷歌趋势、知识图谱）之间的桥梁。

建筑

┌──────────────────────────────────────────────────────────────────┐
│                        MCP Gateway                                │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│   [orchestrator] ─── MCP Protocol ───> [mcp-gateway]             │
│                                              │                    │
│                          ┌───────────────────┼───────────────────┐
│                          │                   │                   │
│                          ▼                   ▼                   ▼
│                   ┌─────────────┐   ┌─────────────┐   ┌─────────┐
│                   │   Google    │   │   YouTube   │   │  Mock   │
│                   │   Trends    │   │  Data API   │   │Provider │
│                   └─────────────┘   └─────────────┘   └─────────┘
│                                                                   │
│   Features:                                                       │
│   • 19 MCP Tools (trends, safety, publishing, analytics)        │
│   • Mock Mode for development without API keys                   │
│   • OAuth2 authentication with token refresh                     │
│   • Circuit breaker pattern for resilience                       │
│   • Rate limiting with async-safe token bucket                   │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

快速开始

# Navigate to the gateway directory
cd mcp-gateway/mcp-gateway

# Install dependencies
uv sync

# Run in mock mode (no API keys required)
MOCK_MODE=true uv run python -m src

# Run with real APIs (requires credentials)
uv run python -m src

可用工具（共19个）

内容发现和趋势

工具	说明	示例使用
`get_trending_topics`	获取具有权威评分的热门话题	查找病毒式传播的内容创意
`search_facts`	通过知识图谱进行事实核查	在编写脚本之前验证声明

出版与管理

工具	说明	示例使用
`publish_video`	以安全的发布流程上传视频	自动上传YouTube
`manage_comments`	评论审核自动化	过滤垃圾邮件，pin突出显示
`get_optimal_publish_time`	智能日程安排建议	最大限度地提高初始参与度

内容安全

工具	说明	示例使用
`check_content_safety`	三层内容过滤	飞行前安全检查
`check_regional_safety`	文化敏感性验证	多市场内容
`check_compliance`	YouTube 2026人工智能披露合规性	避免政策罢工

变现

工具	说明	示例使用
`get_ad_suitability_score`	预测货币化资格	CPM优化
`get_ad_friendly_suggestions`	品牌安全关键字替代方案	提高广告收入
`extract_affiliate_links`	发现联盟机会	收入多样化
`get_monetization_dashboard`	收入情报概述	跟踪收益
`get_metric_details`	深入了解具体指标	分析绩效

搜索引擎优化和人工智能概述

工具	说明	示例使用
`check_aio_status`	查看谷歌人工智能概述归因	SEO可见性
`get_aio_optimization_feedback`	常见问题解答优化建议	提高AI引用

分析与系统

工具	说明	示例使用
`get_analytics`	获取视频性能数据	跟踪观看次数
`get_system_status`	健康检查和断路器状态	运行监测

使用示例

示例1：获取热门话题

# Test trending topics in mock mode
cd mcp-gateway/mcp-gateway
MOCK_MODE=true uv run python -c "
import asyncio
from src.server import mcp

async def test():
    result = await mcp._tool_manager._tools['get_trending_topics'].fn(
        category='technology',
        geo='US',
        max_results=5
    )
    print('Trending Topics:')
    for topic in result.get('topics', []):
        print(f\"  • {topic['keyword']} (authority: {topic['authority_score']})\"
    )
    return result

asyncio.run(test())
"

预期产量：

Trending Topics:
  • AI Code Assistants (authority: 85)
  • Quantum Computing Breakthrough (authority: 72)
  • GPT-5 Release (authority: 91)
  • Rust Programming (authority: 68)
  • WebAssembly Future (authority: 64)

示例2：检查内容安全

cd mcp-gateway/mcp-gateway
MOCK_MODE=true uv run python -c "
import asyncio
from src.server import mcp

async def test():
    result = await mcp._tool_manager._tools['check_content_safety'].fn(
        title='10 Python Tips for Beginners',
        description='Learn Python faster with these proven techniques',
        tags=['python', 'programming', 'tutorial'],
        language='en'
    )
    print(f\"Safety Level: {result['level']}\")
    print(f\"Ad-Friendly: {result['ad_friendly']}\")
    print(f\"Concerns: {result.get('concerns', 'None')}\")
    return result

asyncio.run(test())
"

预期产量：

Safety Level: ad_friendly
Ad-Friendly: True
Concerns: None

示例3：获得最佳发布时间

cd mcp-gateway/mcp-gateway
MOCK_MODE=true uv run python -c "
import asyncio
from src.server import mcp

async def test():
    result = await mcp._tool_manager._tools['get_optimal_publish_time'].fn(
        content_type='main_video',
        target_audience='US'
    )
    print(f\"Recommended Time: {result['recommended_time']}\")
    print(f\"Day: {result['day_of_week']}\")
    print(f\"Timezone: {result['timezone']}\")
    print(f\"Reasoning: {result['reasoning']}\")
    return result

asyncio.run(test())
"

预期产量：

Recommended Time: 14:00
Day: Tuesday
Timezone: America/Los_Angeles
Reasoning: Peak engagement for tech content in US market

示例4：检查系统状态

cd mcp-gateway/mcp-gateway
MOCK_MODE=true uv run python -c "
import asyncio
import json
from src.server import mcp

async def test():
    result = await mcp._tool_manager._tools['get_system_status'].fn()
    print(json.dumps(result, indent=2, default=str))
    return result

asyncio.run(test())
"

预期产量：

{
  "status": "healthy",
  "mock_mode": true,
  "tools_registered": 19,
  "circuit_breakers": {
    "youtube": "closed",
    "trends": "closed"
  },
  "uptime_seconds": 12.5
}

示例5：完整内容验证流程

cd mcp-gateway/mcp-gateway
MOCK_MODE=true uv run python -c "
import asyncio
from src.server import mcp

async def full_validation():
    title = 'Why Rust is Taking Over C++'
    description = 'A deep dive into memory safety and performance'
    tags = ['rust', 'programming', 'cpp', 'memory-safety']

    # Step 1: Content Safety
    safety = await mcp._tool_manager._tools['check_content_safety'].fn(
        title=title,
        description=description,
        tags=tags,
        language='en'
    )
    print(f'1. Safety Check: {safety[\"level\"]}')

    # Step 2: Regional Safety
    regional = await mcp._tool_manager._tools['check_regional_safety'].fn(
        title=title,
        description=description,
        target_regions=['US', 'DE', 'JP']
    )
    print(f'2. Regional Safety: {regional[\"status\"]}')

    # Step 3: Compliance Check
    compliance = await mcp._tool_manager._tools['check_compliance'].fn(
        content_type='educational',
        ai_generated=False,
        sponsored=False
    )
    print(f'3. Compliance: {compliance[\"compliant\"]}')

    # Step 4: Ad Suitability
    ad_score = await mcp._tool_manager._tools['get_ad_suitability_score'].fn(
        title=title,
        description=description,
        category='technology'
    )
    print(f'4. Ad Suitability: {ad_score[\"score\"]}/100')

    # Step 5: Optimal Publish Time
    publish_time = await mcp._tool_manager._tools['get_optimal_publish_time'].fn(
        content_type='main_video',
        target_audience='US'
    )
    print(f'5. Best Publish Time: {publish_time[\"recommended_time\"]} {publish_time[\"timezone\"]}')

    print('\\n✅ Content validated and ready for production!')

asyncio.run(full_validation())
"

预期产量：

1. Safety Check: ad_friendly
2. Regional Safety: approved
3. Compliance: True
4. Ad Suitability: 87/100
5. Best Publish Time: 14:00 America/Los_Angeles

✅ Content validated and ready for production!

配置

环境变量

创建 .env 在 mcp-gateway/mcp-gateway/:

# ===========================================
# MOCK MODE - Enable for development
# ===========================================
MOCK_MODE=true    # Set to 'false' for production

# ===========================================
# Google API Keys (required for production)
# ===========================================
GOOGLE_API_KEY=your_google_api_key_here
GOOGLE_CSE_ID=your_custom_search_engine_id_here

# OAuth2 Client Secrets
GOOGLE_CLIENT_SECRETS_FILE=./credentials/client_secrets.json

# ===========================================
# Directories
# ===========================================
CREDENTIALS_DIR=./credentials
CACHE_DIR=./.cache

# ===========================================
# Logging
# ===========================================
LOG_LEVEL=INFO    # DEBUG | INFO | WARNING | ERROR

# ===========================================
# Rate Limiting (optional)
# ===========================================
RATE_LIMIT_RPM=60

设置真正的API凭据

创建谷歌云项目

# Go to https://console.cloud.google.com
# Create new project: "yt-factory"

启用所需的API
- YouTube数据API v3
- YouTube分析API
- 自定义搜索API
- 知识图谱搜索API

创建OAuth 2.0凭据

# Download client_secrets.json
mkdir -p mcp-gateway/mcp-gateway/credentials
mv ~/Downloads/client_secrets.json mcp-gateway/mcp-gateway/credentials/

首次OAuth流

cd mcp-gateway/mcp-gateway
MOCK_MODE=false uv run python -c "
from src.services.auth_manager import PreemptiveAuthManager
auth = PreemptiveAuthManager()
creds = auth.get_credentials('youtube')
print('✅ YouTube credentials obtained successfully')
"
# Browser will open for OAuth consent

项目结构

mcp-gateway/
└── mcp-gateway/
    ├── src/
    │   ├── __init__.py
    │   ├── __main__.py           # Entry point
    │   ├── server.py             # MCP server with 19 tools
    │   ├── config.py             # Configuration & validation
    │   └── services/
    │       ├── auth_manager.py   # OAuth2 with token refresh
    │       ├── rate_limiter.py   # Async-safe token bucket
    │       ├── mock_provider.py  # Mock responses for dev
    │       ├── youtube_client.py # YouTube API wrapper
    │       └── trends_client.py  # Google Trends wrapper
    ├── credentials/              # OAuth tokens (gitignored)
    ├── .cache/                   # API response cache
    ├── pyproject.toml
    └── .env

模拟模式

Mock模式为没有API证书的开发提供了真实的模拟响应：

# Enable mock mode
export MOCK_MODE=true

# Or set in .env
echo "MOCK_MODE=true" >> .env

# Run with mock mode
uv run python -m src

模拟模式功能：

具有权威评分的现实趋势主题
模拟内容安全响应
使用假视频ID上传模拟视频
模拟分析数据
一致、可重复的响应

与编排器集成

MCP网关设计为由编排器通过MCP协议调用：

// In orchestrator, the gateway is configured in .env:
MCP_GATEWAY_COMMAND=uv run python -m src
MCP_GATEWAY_CWD=../mcp-gateway/mcp-gateway

// The orchestrator calls tools like:
const trends = await mcpClient.call('get_trending_topics', {
  category: 'technology',
  geo: 'US',
  max_results: 10
});

技术栈

组件	技术
运行时	Python 3.11+
包装管理器	UV
MCP框架	MCP-python
HTTP客户端	httpx（异步）
OAuth	谷歌认证oauthlib
验证	Pydantic
缓存	磁盘缓存

故障排除

常见问题

“MOCK_MODE不工作”

# Set directly in command
MOCK_MODE=true uv run python -m src

“OAuth令牌已过期”

# Delete cached tokens
rm -f credentials/token_*.json
# Re-run to trigger OAuth flow

“超出费率限制”

# Check circuit breaker status
MOCK_MODE=true uv run python -c "
import asyncio
from src.server import mcp
asyncio.run(mcp._tool_manager._tools['get_system_status'].fn())
"

“导入错误”

# Reinstall dependencies
cd mcp-gateway/mcp-gateway
rm -rf .venv uv.lock
uv sync

最近更新（2026年2月）

关键Bug修复：修复了auth_manager和rate_limiter中的异步阻塞问题
配置验证：已添加 validate_config() 获取所需凭据
速率限制：实施抖动以防止羊群打雷
错误处理：改进了错误消息和日志记录

部分 YT工厂 YouTube自动化生态系统

使用说明

MCP网关

建筑

快速开始

可用工具（共19个）

内容发现和趋势

出版与管理

内容安全

变现

搜索引擎优化和人工智能概述

分析与系统

使用示例

示例1：获取热门话题

示例2：检查内容安全

示例3：获得最佳发布时间

示例4：检查系统状态

示例5：完整内容验证流程

配置

环境变量

设置真正的API凭据

项目结构

模拟模式

与编排器集成

技术栈

故障排除

常见问题

最近更新（2026年2月）

相关 MCP

YouTube AI代理工具

版权保护服务

YouTube自动化代理

模型上下文协议服务

ADR 分析服务器

MCP网关工具