Scriptone

python -m pip install fastapi pydantic todoist-api-python fastapi-mcp

from datetime import datetime
from zoneinfo import ZoneInfo

from fastapi import APIRouter

router = APIRouter(prefix="/utils", tags=["utils"])


@router.post("/current_time", operation_id="current_time", response_model=None)
async def get_current_time() -> datetime:
    return datetime.now(tz=ZoneInfo("Asia/Tokyo"))

import logging
import os
from datetime import date, datetime, timedelta
from typing import Annotated, Literal

import fastapi
from fastapi import APIRouter
from pydantic import BaseModel, Field
from todoist_api_python.api import TodoistAPI
from todoist_api_python.models import Task

today = date.today()  # noqa: DTZ011
tomorrow = today + timedelta(days=1)
yesterday = today - timedelta(days=1)
this_week = today + timedelta(days=7)
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
todoist_api_key = os.environ.get("TODOIST_API_KEY", "")
PROJECT_ID = "xxxxxxxxx"


class TaskParam(BaseModel):
    """Todoistタスクのパラメータを定義するモデル。"""
    content: str = Field(description="タスクの名前")
    description: str | None = Field(description="タスクの説明")
    priority: Literal[1, 2, 3, 4] | None = Field(description="タスクの優先度。１が最も重要度が高く、4が最も重要度が低い。指定がない場合は原則としてNoneを設定する。")
    due_datetime: datetime | None = Field(description="タスクの期限")


def _get_tasks() -> list[Task]:
    """
    Todoist APIからすべてのタスクを取得する。

    Returns:
        list[Task]: 取得したタスクのリスト。

    Raises:
        Exception: API呼び出しに失敗した場合。
    """
    try:
        api = TodoistAPI(token=todoist_api_key)
        tasks_objs = api.get_tasks()
        tasks = []
        for tasks_obj in tasks_objs:
            tasks.extend(tasks_obj)
    except Exception as e:
        msg = str(e)
        logger.exception("タスクの取得に失敗しました: %s", msg)
        raise
    else:
        return tasks


def _get_someday_tasks(date: date) -> list[Task]:
    """
    特定の日付に期限が設定されているタスクを取得する。

    Args:
        date (date): タスクの期限日。

    Returns:
        list[Task]: 指定した日付に期限が設定されているタスクのリスト。
    """
    tasks = _get_tasks()
    return [t for t in tasks if t.due is not None and t.due.date == date]


def _get_today_tasks() -> list[Task]:
    """
    今日のタスクを取得する。

    Returns:
        list[Task]: 今日に期限が設定されているタスクのリスト。
    """
    return _get_someday_tasks(today)


def _get_tomorrow_tasks() -> list[Task]:
    """
    明日のタスクを取得する。

    Returns:
        list[Task]: 明日に期限が設定されているタスクのリスト。
    """
    return _get_someday_tasks(tomorrow)


def _get_yesterday_tasks() -> list[Task]:
    """
    昨日のタスクを取得する。

    Returns:
        list[Task]: 昨日に期限が設定されているタスクのリスト。
    """
    return _get_someday_tasks(yesterday)


def _get_this_week_tasks() -> list[Task]:
    """
    今週のタスクを取得する。

    Returns:
        list[Task]: 今週に期限が設定されているタスクのリスト。
    """
    tasks = _get_tasks()
    return [t for t in tasks if t.due is not None and this_week >= t.due.date >= today]


router = APIRouter(prefix="/todoist", tags=["todoist"])


@router.post("/get_today_tasks", operation_id="get_today_tasks", response_model=None)
async def get_today_tasks() -> list[Task]:
    """
    Todoistから今日のタスクを取得するエンドポイント。

    Returns:
        list[Task]: 今日のタスクのリスト。

    Raises:
        HTTPException: API呼び出しに失敗した場合。
    """
    try:
        return _get_today_tasks()
    except Exception as e:
        msg = str(e)
        logger.exception("今日のタスク取得エンドポイントでエラー: %s", msg)
        raise fastapi.HTTPException(status_code=500, detail=str(e)) from e


@router.post("/get_tomorrow_tasks", operation_id="get_tomorrow_tasks", response_model=None)
async def get_tomorrow_tasks() -> list[Task]:
    """
    Todoistから明日のタスクを取得するエンドポイント。

    Returns:
        list[Task]: 明日のタスクのリスト。

    Raises:
        HTTPException: API呼び出しに失敗した場合。
    """
    try:
        return _get_tomorrow_tasks()
    except Exception as e:
        msg = str(e)
        logger.exception("明日のタスク取得エンドポイントでエラー: %s", msg)
        raise fastapi.HTTPException(status_code=500, detail=str(e)) from e


@router.post("/get_yesterday_tasks", operation_id="get_yesterday_tasks", response_model=None)
async def get_yesterday_tasks() -> list[Task]:
    """
    Todoistから昨日のタスクを取得するエンドポイント。

    Returns:
        list[Task]: 昨日のタスクのリスト。

    Raises:
        HTTPException: API呼び出しに失敗した場合。
    """
    try:
        return _get_yesterday_tasks()
    except Exception as e:
        msg = str(e)
        logger.exception("昨日のタスク取得エンドポイントでエラー: %s", msg)
        raise fastapi.HTTPException(status_code=500, detail=str(e)) from e


@router.post("/get_this_week_tasks", operation_id="get_this_week_tasks", response_model=None)
async def get_this_week_tasks() -> list[Task]:
    """
    Todoistから今週のタスクを取得するエンドポイント。

    Returns:
        list[Task]: 今週のタスクのリスト。

    Raises:
        HTTPException: API呼び出しに失敗した場合。
    """
    try:
        return _get_this_week_tasks()
    except Exception as e:
        msg = str(e)
        logger.exception("今週のタスク取得エンドポイントでエラー: %s", msg)
        raise fastapi.HTTPException(status_code=500, detail=str(e)) from e


@router.post(
    "/put_task",
    operation_id="put_task",
)
async def put_task(task_param: Annotated[TaskParam, fastapi.Body(embed=True)]) -> dict[str, str]:
    """
    Todoistに新しいタスクを追加するエンドポイント。

    Args:
        task_param (TaskParam): 追加するタスクのパラメータ。

    Returns:
        dict[str, str]: 処理の結果。成功した場合はステータスと成功メッセージ、失敗した場合はエラーメッセージを含む。
    """
    try:
        api = TodoistAPI(token=todoist_api_key)
        api.add_task(
            content=task_param.content,
            description=task_param.description,
            priority=task_param.priority,
            project_id=PROJECT_ID,
            due_datetime=task_param.due_datetime,
        )
        logger.info("タスク '%s' を追加しました。", task_param.content)

    except Exception as e:
        msg = str(e)
        logger.exception("タスク追加エンドポイントでエラー: %s", msg)
        return {"status": "error", "message": msg}
    else:
        return {"status": "success", "message": "タスクが正常に追加されました。"}


@router.post(
    "/put_tasks",
    operation_id="put_tasks",
)
async def put_tasks(task_params: Annotated[list[TaskParam], fastapi.Body(embed=True)]) -> dict[str, list[dict[str, str]]]:
    """
    Todoistに複数の新しいタスクを追加するエンドポイント。

    Args:
        task_params (list[TaskParam]): 追加するタスクのパラメータのリスト。

    Returns:
        dict[str, list[dict[str, str]]]: 各タスクの処理結果のリスト。
    """
    results: list[dict[str, str]] = []
    for task_param in task_params:
        r = await put_task(task_param)
        results.append(r)
    logger.info("%s件のタスク追加処理が完了しました。", str(len(results)))
    return {"results": results}

import fastapi

from . import todoist, utils

app = fastapi.FastAPI(
    title="MCP API",
    description="API for managing the MCP system",
    version="0.1.0",
)
app.include_router(todoist.router)
app.include_router(utils.router)

import os

from fastapi_mcp import FastApiMCP

from libs import api

port = int(os.environ.get("PORT", "8000"))

mcp = FastApiMCP(
    api.app,
    name="My API MCP",
    description="My API description",
    # base_url="https://example.com",
)

mcp.mount()

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(api.app, host="0.0.0.0", port=port)  # noqa: S104

uv tool install mcp-proxy

{
  "command": "mcp-proxy",
  "args": [
    "https://example.com/mcp"
  ],
  "env": {}
}