JinDaGe - mytaskly-mcp MCP Details

article

README

🚀 MyTaskly MCP 服务器

MyTaskly MCP 服务器是为 MyTaskly 打造的 模型上下文协议（MCP）服务器，具备 OAuth 2.1 JWT 认证 功能，并能与 FastAPI 后端实现无缝集成。

✨ 主要特性

🔐 企业级认证

OAuth 2.1 JWT：遵循 MCP 2025 标准（RFC 8707）的安全令牌认证方式。
多用户支持：通过 JWT 令牌验证，单个部署即可服务所有用户。
受众声明验证：防止令牌在不同服务间重复使用。

🚀 高性能集成

HTTP API 网关：与 FastAPI 后端通信，无需直接访问数据库。
无状态架构：无需会话管理，具备完全可扩展性。
连接池：优化后的 HTTP 客户端，实现高吞吐量。

📱 移动优先设计

React Native 优化：返回的数据格式适配原生移动组件。
语音友好响应：包含用于聊天应用 TTS 的语音摘要。
预格式化 UI 数据：包含表情符号、颜色和格式化日期，可直接显示。

🛠️ 可用的 MCP 工具（共 20 个）

MCP 服务器提供 20 个工具，分为 5 个类别，用于全面的任务管理。

📋 任务工具（8 个）

| 工具 | 描述 | 是否需要认证 | |------|-------------|---------------| | get_tasks | 获取带有过滤器的任务（为 React Native 格式化） | ✅ 是 | | add_task | 创建新任务，具备智能分类处理功能 | ✅ 是 | | update_task | 更新任务字段 | ✅ 是 | | complete_task | 快速将任务标记为已完成 | ✅ 是 | | get_task_stats | 获取任务统计信息（总数、已完成数、按优先级统计） | ✅ 是 | | get_next_due_task | 获取即将到期的 N 个任务 | ✅ 是 | | get_overdue_tasks | 获取所有逾期任务 | ✅ 是 | | get_upcoming_tasks | 获取未来 N 天内到期的任务 | ✅ 是 |

示例响应 - get_tasks:

{
  "type": "task_list",
  "tasks": [
    {
      "id": 123,
      "title": "Pizza",
      "endTimeFormatted": "Venerdì 15 dicembre, 18:00",
      "category": "Cibo",
      "categoryColor": "#EF4444",
      "priority": "Alta",
      "priorityEmoji": "⚡",
      "status": "Pending",
      "actions": {
        "canEdit": true,
        "canDelete": true,
        "canComplete": true
      }
    }
  ],
  "summary": {
    "total": 10,
    "pending": 5,
    "completed": 3,
    "high_priority": 2
  },
  "voice_summary": "Hai 10 task, di cui 2 ad alta priorità. 5 sono in sospeso e 3 completati."
}

📂 分类工具（4 个）

| 工具 | 描述 | 是否需要认证 | |------|-------------|---------------| | get_my_categories | 获取所有用户分类 | ✅ 是 | | create_category | 创建新分类 | ✅ 是 | | update_category | 按 ID 更新分类 | ✅ 是 | | search_categories | 通过模糊匹配搜索分类 | ✅ 是 |

示例响应 - get_my_categories:

{
  "categories": [
    {
      "category_id": 1,
      "name": "Lavoro",
      "description": "Task di lavoro",
      "is_shared": true,
      "owner_id": 1,
      "permission_level": "READ_WRITE"
    }
  ],
  "total": 5,
  "owned": 3,
  "shared_with_me": 2
}

📝 笔记工具（4 个）

| 工具 | 描述 | 是否需要认证 | |------|-------------|---------------| | get_notes | 获取所有用户笔记 | ✅ 是 | | create_note | 创建新笔记（便签风格） | ✅ 是 | | update_note | 更新笔记的文本、位置或颜色 | ✅ 是 | | delete_note | 删除笔记 | ✅ 是 |

示例响应 - create_note:

{
  "note_id": 456,
  "title": "Comprare il latte",
  "color": "#FFEB3B",
  "position_x": 100.5,
  "position_y": 250.0,
  "created_at": "2025-01-15T10:30:00Z",
  "message": "✅ Nota creata con successo"
}

🔧 元工具（3 个）

| 工具 | 描述 | 是否需要认证 | |------|-------------|---------------| | get_or_create_category | 通过模糊匹配智能查找或创建分类 | ✅ 是 | | move_all_tasks_between_categories | 在不同分类之间批量移动任务 | ✅ 是 | | add_multiple_tasks | 一次性批量创建多个任务 | ✅ 是 |

⚕️ 系统工具（1 个）

| 工具 | 描述 | 是否需要认证 | |------|-------------|---------------| | health_check | 检查服务器健康状况和连接性 | ❌ 否 |

示例响应 - health_check:

{
  "mcp_server": "healthy",
  "fastapi_server": "healthy",
  "fastapi_url": "http://localhost:8080",
  "timestamp": "2025-01-15T10:30:00Z",
  "version": "2.0.0"
}

🚀 快速开始

使用选项

你有 两种方式 来使用 MyTaskly MCP 服务器：

选项 1：使用官方公共服务器（推荐）

使用 官方 MyTaskly MCP 服务器（即将推出），无需进行任何设置！

# 配置你的 MCP 客户端连接到：
# https://mcp.mytasklyapp.com （URL 即将发布）

优点：

✅ 无需安装或配置
✅ 始终保持最新功能
✅ 由 MyTaskly 团队管理和监控
✅ 可直接与 MyTaskly 移动应用配合使用

选项 2：自行搭建（高级用户）

运行你自己的本地 MCP 服务器实例。

前提条件：

Python 3.11+（建议使用虚拟环境）
本地运行的 MyTaskly FastAPI 服务器（详见 MyTaskly-server）
与你的 FastAPI 服务器配置匹配的 JWT 密钥
配置为使用自定义服务器的 修改后的 MyTaskly 应用

快速开始（5 分钟）：

git clone https://github.com/Gabry848/MyTaskly-mcp.git
cd MyTaskly-mcp
python -m venv venv && pip install -r requirements.txt
cp .env.example .env && python main.py

⚠️ 重要提示： 自行搭建时，你还必须：

运行本地的 MyTaskly-server 实例。
修改 MyTaskly 移动应用，使其指向你的自定义服务器 URL。

自行搭建设置指南

1. 克隆并安装

# 克隆仓库
git clone https://github.com/Gabry848/MyTaskly-mcp.git
cd MyTaskly-mcp

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate

# 安装依赖
pip install -r requirements.txt

2. 配置环境变量

在根目录下创建 .env 文件：

# ============ FASTAPI 后端 ============
FASTAPI_BASE_URL=http://localhost:8080
FASTAPI_API_KEY=your_api_key_here

# ============ JWT 配置 ============
# 关键：必须与 FastAPI 服务器配置匹配！
JWT_SECRET_KEY=your_jwt_secret_key_here
JWT_ALGORITHM=HS256
MCP_AUDIENCE=mytaskly-mcp

# ============ 服务器配置 ============
MCP_SERVER_NAME=MyTaskly-MCP
MCP_SERVER_VERSION=2.0.0
LOG_LEVEL=INFO

⚠️ 关键提示： JWT_SECRET_KEY 必须与你的 FastAPI 服务器的 SECRET_KEY 环境变量匹配！

3. 启动 MCP 服务器

python main.py

服务器将以 标准输入输出模式 启动，并显示可用工具。配置你的 MCP 客户端连接到该服务器。

🔐 认证与安全

OAuth 2.1 流程

MCP 服务器使用遵循 OAuth 2.1 和 RFC 8707 标准的 JWT 令牌：

┌─────────────────┐
│  移动客户端  │
│  (React Native) │
└────────┬────────┘
         │ 1. 登录请求
         ▼
┌─────────────────┐
│  FastAPI 服务器 │  2. 验证凭据
│  (认证服务器)  │  3. 生成带有 MCP 受众声明的 JWT
└────────┬────────┘
         │ 4. 返回 JWT 令牌
         ▼
┌─────────────────┐
│  移动客户端  │  5. 安全存储令牌
└────────┬────────┘
         │ 6. 使用授权头调用 MCP 工具
         ▼
┌─────────────────┐
│   MCP 服务器    │  7. 验证 JWT 签名
│ (本项目)  │  8. 验证受众声明
│                 │  9. 从令牌中提取用户 ID
└────────┬────────┘
         │ 10. 使用用户 ID 向 FastAPI 发送 HTTP 请求
         ▼
┌─────────────────┐
│  FastAPI 服务器 │  11. 返回特定用户的数据
│ (资源 API)  │
└────────┬────────┘
         │ 12. 格式化数据以适配移动 UI
         ▼
┌─────────────────┐
│   MCP 服务器    │  13. 返回格式化响应
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  移动客户端  │  14. 渲染 UI / 播放 TTS
└─────────────────┘

JWT 令牌结构

JWT 必须包含以下声明（遵循 RFC 7519 和 RFC 8707）：

{
  "sub": "123",                              // 用户 ID（必需）
  "aud": "mcp://mytaskly-mcp-server",       // 受众（必需，RFC 8707）
  "iss": "https://api.mytasklyapp.com",     // 颁发者（可选）
  "exp": 1735689600,                         // 过期时间戳（必需）
  "iat": 1735686000,                         // 颁发时间戳（必需）
  "scope": "tasks:read tasks:write notes:write" // 权限范围（可选）
}

安全特性： | 特性 | 实现方式 | |---------|----------------| | 签名验证 | 使用共享密钥的 HS256 算法 | | 受众声明 | 防止令牌在不同服务间重复使用 | | 过期检查 | 自动使过期令牌失效 | | 用户隔离 | 每个请求都限定于经过认证的用户 |

获取 JWT 令牌

选项 1：从 FastAPI 获取（生产环境）

你需要在 FastAPI 服务器中添加以下端点：

# src/app/api/routes/auth.py

@router.post("/auth/mcp-token")
async def get_mcp_token(current_user: User = Depends(get_current_user)):
    """为 MCP 服务器访问生成 JWT 令牌。"""
    payload = {
        "sub": str(current_user.user_id),
        "aud": "mcp://mytaskly-mcp-server",
        "iss": "https://api.mytasklyapp.com",
        "exp": datetime.utcnow() + timedelta(minutes=30),
        "iat": datetime.utcnow(),
        "scope": "tasks:read tasks:write categories:read notes:read notes:write"
    }
    token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
    return {"mcp_token": token, "expires_in": 1800}

选项 2：生成测试令牌（开发环境）

from src.auth import create_test_token

# 为用户 ID 为 1 的用户生成测试令牌
token = create_test_token(user_id=1, expires_minutes=30)
print(f"测试令牌: {token}")

🧪 测试与开发

使用 Python 进行手动测试

import asyncio
from src.auth import create_test_token
from src.server import get_tasks, get_categories, create_note

async def test_mcp_tools():
    """使用生成的令牌测试所有 MCP 工具。"""

    # 为用户 ID 为 1 的用户生成测试令牌（30 分钟后过期）
    token = create_test_token(user_id=1, expires_minutes=30)
    auth_header = f"Bearer {token}"

    print("🔑 为用户 ID 为 1 的用户生成了测试令牌\n")

    # 测试 1: 获取任务
    print("1️⃣ 测试 get_tasks...")
    tasks = await get_tasks(authorization=auth_header)
    print(f"   ✅ 检索到 {tasks['summary']['total']} 个任务")
    print(f"   📊 摘要: {tasks['summary']}")
    print(f"   🎤 语音: {tasks['voice_summary']}\n")

    # 测试 2: 获取分类
    print("2️⃣ 测试 get_categories...")
    categories = await get_categories(authorization=auth_header)
    print(f"   ✅ 检索到 {categories['total']} 个分类")
    print(f"   📂 拥有的: {categories.get('owned', 0)}")
    print(f"   🤝 共享的: {categories.get('shared_with_me', 0)}\n")

    # 测试 3: 创建笔记
    print("3️⃣ 测试 create_note...")
    note = await create_note(
        authorization=auth_header,
        title="Test note from MCP",
        color="#4CAF50",
        position_x=100.0,
        position_y=200.0
    )
    print(f"   ✅ 创建了笔记 #{note['note_id']}")
    print(f"   📝 标题: {note['title']}")
    print(f"   🎨 颜色: {note['color']}\n")

    print("✅ 所有测试均成功完成！")

# 运行测试
if __name__ == "__main__":
    asyncio.run(test_mcp_tools())

使用 cURL 进行测试

# 1. 生成测试 JWT 令牌
python -c "from src.auth import create_test_token; print(create_test_token(1))"

# 2. 将令牌导出到环境变量（替换为实际令牌）
export MCP_TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

# 3. 测试 get_tasks
curl -X POST http://localhost:8000/mcp/get_tasks \
  -H "Authorization: Bearer $MCP_TOKEN" \
  -H "Content-Type: application/json"

# 4. 测试 get_categories
curl -X POST http://localhost:8000/mcp/get_categories \
  -H "Authorization: Bearer $MCP_TOKEN" \
  -H "Content-Type: application/json"

# 5. 测试 create_note
curl -X POST http://localhost:8000/mcp/create_note \
  -H "Authorization: Bearer $MCP_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Meeting notes",
    "color": "#FF5722",
    "position_x": 150.5,
    "position_y": 300.0
  }'

# 6. 测试 health_check（无需认证）
curl -X GET http://localhost:8000/mcp/health_check

自动化测试套件

# 运行所有单元测试
python -m pytest tests/ -v

# 运行特定测试文件
python -m pytest tests/test_auth.py -v

# 运行并生成覆盖率报告
python -m pytest tests/ --cov=src --cov-report=html

# 运行并输出结果
python -m pytest tests/ -v -s

💻 使用示例

与 React Native 集成

get_tasks 工具返回的数据经过优化，适用于 React Native 组件：

import { FlatList, View, Text } from 'react-native';

async function fetchTasks() {
  // 从你的认证系统获取 JWT 令牌
  const token = await getAuthToken();

  // 调用 MCP 服务器
  const response = await mcpClient.call('get_tasks', {
    authorization: `Bearer ${token}`
  });

  return response;
}

function TasksList() {
  const [data, setData] = useState(null);

  useEffect(() => {
    fetchTasks().then(setData);
  }, []);

  if (!data) return <Loading />;

  return (
    <View>
      {/* 语音摘要，提高可访问性 */}
      <Text accessible>{data.voice_summary}</Text>

      {/* 渲染任务列表 */}
      <FlatList
        data={data.tasks}
        renderItem={({ item }) => (
          <TaskCard
            title={item.title}
            date={item.endTimeFormatted}
            category={item.category}
            categoryColor={item.categoryColor}
            priority={item.priorityEmoji}
          />
        )}
      />
    </View>
  );
}

与语音聊天集成

响应中包含用于 TTS 的 voice_summary：

# 在你的聊天机器人服务中
response = await mcp_client.call('get_tasks', {
    'authorization': f'Bearer {user_jwt}'
})

# 用于可视化显示
ui_data = response['tasks']

# 用于语音输出
tts_text = response['voice_summary']
# "Hai 10 task, di cui 2 ad alta priorità. 5 sono in sospeso e 3 completati."

🔒 安全最佳实践

生产环境始终使用 HTTPS。
确保 JWT_SECRET_KEY 安全：切勿将其提交到 Git。
使用短期令牌（15 - 30 分钟）。
在客户端实现令牌刷新机制。
验证受众声明（RFC 8707）：防止令牌重复使用。
记录认证失败信息 以便监控。

🏗️ 架构与项目结构

系统架构

┌─────────────────────────────────────────────────────────────┐
│                    MyTaskly 生态系统                        │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│   ┌─────────────────┐                                       │
│   │  移动客户端  │  1. 用户认证               │
│   │ (React Native)  │  2. 接收 JWT 令牌                │
│   └────────┬────────┘  3. 调用 MCP 工具                   │
│            │                                                  │
│            ▼                                                  │
│   ┌─────────────────┐                                       │
│   │   MCP 服务器    │  4. 验证 JWT（OAuth 2.1）        │
│   │ (本项目)  │  5. 从令牌中提取用户 ID       │
│   └────────┬────────┘  6. 格式化数据以适配移动 UI        │
│            │                                                  │
│            ▼                                                  │
│   ┌─────────────────┐                                       │
│   │  FastAPI 服务器 │  7. 处理业务逻辑            │
│   │ (MyTaskly-API)  │  8. 管理数据库操作       │
│   └────────┬────────┘  9. 返回原始数据                  │
│            │                                                  │
│            ▼                                                  │
│   ┌─────────────────┐                                       │
│   │   PostgreSQL    │  10. 持久化存储               │
│   │    数据库     │  11. 触发器和通知         │
│   └─────────────────┘                                       │
│                                                               │
└─────────────────────────────────────────────────────────────┘

项目结构

MyTaskly-mcp/
├── src/
│   ├── core/                      # 核心 MCP 服务器
│   │   ├── __init__.py
│   │   └── server.py             # FastMCP 实例和工具注册
│   │
│   ├── client/                    # HTTP 客户端层
│   │   ├── __init__.py
│   │   ├── base.py               # 带有认证的基础 HTTP 客户端
│   │   ├── categories.py         # 分类 API 端点
│   │   ├── tasks.py              # 任务 API 端点
│   │   ├── notes.py              # 笔记 API 端点
│   │   └── health.py             # 健康检查端点
│   │
│   ├── tools/                     # MCP 工具（业务逻辑）
│   │   ├── __init__.py
│   │   ├── categories.py         # 分类工具（4 个方法）
│   │   ├── tasks.py              # 任务工具（8 个方法）
│   │   ├── notes.py              # 笔记工具（4 个方法）
│   │   ├── meta.py               # 元工具（3 个方法）
│   │   └── health.py             # 健康检查工具（1 个方法）
│   │
│   ├── formatters/                # 响应格式化器
│   │   ├── __init__.py
│   │   └── tasks.py              # 为 React Native UI 格式化任务数据
│   │
│   ├── auth.py                    # JWT 认证
│   ├── config.py                  # 配置设置
│   └── http_server.py            # 可选的 HTTP 服务器包装器
│
├── tests/                         # 测试套件
├── main.py                        # 主入口点
├── pyproject.toml                 # 项目配置
├── requirements.txt               # Python 依赖项
├── ARCHITECTURE.md                # 详细架构文档
└── README.md                      # 本文件

分层架构

| 层 | 文件 | 职责 | |-------|-------|----------------| | 核心层 | src/core/ | MCP 服务器实例和工具注册 | | 工具层 | src/tools/ | 定义带有业务逻辑的 MCP 工具（共 20 个工具） | | 客户端层 | src/client/ | 与 FastAPI 服务器进行 HTTP 通信 | | 格式化器层 | src/formatters/ | 转换 API 响应以适配 React Native UI |

关键组件

| 组件 | 技术 | 用途 | |-----------|-----------|---------| | MCP 服务器 | 带有 asyncio 的 FastMCP | 请求处理和工具编排 | | JWT 认证 | 带有 HS256 的 PyJWT | 基于令牌的安全认证 | | HTTP 客户端 | httpx（异步） | 与 FastAPI 后端通信 | | 数据格式化 | 自定义格式化器 | 优化响应结构以适配移动设备 |

📚 如需详细架构信息，请参阅 ARCHITECTURE.md

🛠️ 开发指南

添加新的 MCP 工具

遵循分层架构模式：

1. 添加 HTTP 客户端方法

# src/client/tasks.py
async def new_operation(self, user_id: int, params...) -> Dict[str, Any]:
    """调用新的 FastAPI 端点。"""
    token = await self._get_user_token(user_id)
    return await self._post("/new-endpoint", token, json={...})

2. 添加 MCP 工具

# src/tools/tasks.py
async def new_tool(authorization: str, params...) -> Dict[str, Any]:
    """工具文档说明。"""
    user_id = verify_jwt_token(authorization)
    result = await task_client.new_operation(user_id, params)
    return format_response(result)

3. 注册工具

# src/core/server.py
from src.tools.tasks import new_tool
mcp.tool()(new_tool)

4. 更新 main.py 横幅

在 print_banner() 中添加新工具到列表。

更多详细信息，请参阅 ARCHITECTURE.md

代码质量

# 使用 Black 格式化代码
black src/ tests/

# 使用 Ruff 进行代码检查
ruff check src/ tests/

# 使用 mypy 进行类型检查
mypy src/

# 运行测试并生成覆盖率报告
pytest tests/ --cov=src --cov-report=html

常见开发任务

| 任务 | 命令 | |------|---------| | 运行服务器 | python main.py | | 生成测试令牌 | python -c "from src.auth import create_test_token; print(create_test_token(1))" | | 运行测试 | pytest tests/ -v | | 检查覆盖率 | pytest tests/ --cov=src | | 格式化代码 | black src/ tests/ | | 安装依赖 | pip install -r requirements.txt |

📚 资源与相关项目

MyTaskly 生态系统

MyTaskly 移动应用：React Native 前端
MyTaskly 服务器：FastAPI 后端
MyTaskly MCP（本项目）：模型上下文协议服务器

文档

🤝 贡献指南

我们欢迎贡献！本项目是 MyTaskly 生态系统的一部分。

如何贡献

** Fork 仓库**
创建功能分支：git checkout -b feature/my-feature
进行更改，并使用清晰的提交信息
为新功能添加测试
确保测试通过：pytest tests/ -v
格式化代码：black src/ tests/
提交拉取请求

开发工作流

# 1. 克隆你的 Fork
git clone https://github.com/YOUR_USERNAME/MyTaskly-mcp.git
cd MyTaskly-mcp

# 2. 创建功能分支
git checkout -b feature/my-feature

# 3. 进行更改并测试
pytest tests/ -v

# 4. 提交带有描述性信息的更改
git commit -m "feat: add new MCP tool for task statistics"

# 5. 推送并创建 PR
git push origin feature/my-feature

📄 许可证

本项目采用 MIT 许可证，详情请参阅 LICENSE 文件。

MIT 许可证允许你：

✅ 商业使用
✅ 修改
✅ 分发
✅ 个人使用

📞 支持与反馈

问题反馈：GitHub Issues
讨论交流：GitHub Discussions
邮件支持：support@mytasklyapp.com

由 Gabry848 用心打造，作为 MyTaskly 项目的一部分

感谢点赞！ ⭐

⬆ 返回顶部