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="작업 우선순위. 1이 가장 높은 우선순위이고, 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": {}
}