【実装】既存REST APIをMCPサーバー化する：Pythonでタスク管理APIをAIから操作する

POST /tasks
GET /tasks
GET /tasks/{task_id}

{
  "title": "MCPの記事を公開する",
  "description": "図とコードを最終確認する",
  "priority": "high",
  "due_date": "2026-07-16"
}

{
  "id": "b48b8da3-f80b-40e3-a78d-8d70b11e58c4",
  "title": "MCPの記事を公開する",
  "description": "図とコードを最終確認する",
  "priority": "high",
  "due_date": "2026-07-16",
  "status": "todo",
  "created_at": "2026-07-15T21:30:00+09:00"
}

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "title": "MCPの記事を公開する",
    "priority": "high",
    "due_date": "2026-07-16"
  }'

git clone https://github.com/yunosuke-github/task-api-mcp.git
cd task-api-mcp
uv sync --dev

[project]
name = "task-api-mcp"
version = "0.1.0"
description = "A local task REST API with a thin MCP adapter"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
dependencies = [
    "fastapi>=0.115,<1",
    "httpx>=0.27,<1",
    "mcp[cli]==1.28.1",
    "pydantic>=2.10,<3",
    "uvicorn>=0.30,<1",
]

[project.scripts]
task-api = "task_api.app:main"
task-mcp = "task_mcp.server:main"

[dependency-groups]
dev = [
    "pytest>=8.3,<9",
    "pytest-asyncio>=0.24,<2",
]

task-api-mcp/
├── data/tasks.json
├── src/
│   ├── task_api/
│   │   ├── app.py
│   │   ├── models.py
│   │   └── repository.py
│   └── task_mcp/
│       ├── api_client.py
│       ├── errors.py
│       ├── models.py
│       └── server.py
├── tests/
├── .env.example
├── pyproject.toml
└── uv.lock

TASK_API_BASE_URL=http://localhost:8000
TASK_API_TIMEOUT_SECONDS=10
TASK_API_DATA_FILE=data/tasks.json

cp .env.example .env
uv run --env-file .env task-api
# 別ターミナルでMCPサーバーを直接起動する場合
uv run --env-file .env task-mcp

"""Application-level errors raised by the task REST API client."""


class TaskApiValidationError(Exception):
    """The API rejected a request that the caller can correct."""


class TaskApiUnavailableError(Exception):
    """The API could not be reached or returned a temporary failure."""


class TaskApiUnexpectedError(Exception):
    """The API returned an unexpected status or response body."""

"""Async HTTP client used by the MCP adapter."""

from typing import Any

import httpx

from .errors import (
    TaskApiUnexpectedError,
    TaskApiUnavailableError,
    TaskApiValidationError,
)
from .models import CreateTaskRequest, TaskResult


class TaskApiClient:
    """Translate HTTP outcomes from the task API into domain-level errors."""

    def __init__(
        self,
        base_url: str,
        timeout_seconds: float,
        transport: httpx.AsyncBaseTransport | None = None,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._timeout = timeout_seconds
        self._transport = transport

    async def create_task(self, request: CreateTaskRequest) -> TaskResult:
        """Create a task through the REST API and validate its response."""

        try:
            async with httpx.AsyncClient(
                base_url=self._base_url,
                timeout=self._timeout,
                transport=self._transport,
            ) as client:
                response = await client.post(
                    "/tasks",
                    json=request.model_dump(mode="json"),
                )
        except (httpx.TimeoutException, httpx.RequestError) as exc:
            raise TaskApiUnavailableError from exc

        if response.status_code == 201:
            try:
                return TaskResult.model_validate(response.json())
            except (ValueError, TypeError) as exc:
                raise TaskApiUnexpectedError from exc

        if response.status_code in {400, 409, 422}:
            raise TaskApiValidationError(self._safe_detail(response))

        if response.status_code >= 500:
            raise TaskApiUnavailableError

        raise TaskApiUnexpectedError

    @staticmethod
    def _safe_detail(response: httpx.Response) -> str:
        """Extract only a bounded, user-correctable API error message."""

        try:
            body: Any = response.json()
        except ValueError:
            return "The task request was rejected."

        detail = body.get("detail") if isinstance(body, dict) else None
        if isinstance(detail, str):
            return detail[:300]
        return "The task request was rejected."

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="task-manager",
    json_response=True,
)

"""Models at the boundary between MCP and the task REST API."""

from datetime import date, datetime
from typing import Literal

from pydantic import BaseModel

Priority = Literal["low", "medium", "high"]


class CreateTaskRequest(BaseModel):
    """JSON body sent to ``POST /tasks``."""

    title: str
    description: str | None = None
    priority: Priority = "medium"
    due_date: date | None = None


class TaskResult(BaseModel):
    """Structured task returned to the MCP client."""

    id: str
    title: str
    description: str | None = None
    priority: Priority
    due_date: date | None = None
    status: str
    created_at: datetime

"""stdio MCP server exposing the task API as a model-friendly tool."""

import logging
import os
import sys
from datetime import date
from typing import Annotated

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from pydantic import Field

from .api_client import TaskApiClient
from .errors import (
    TaskApiUnexpectedError,
    TaskApiUnavailableError,
    TaskApiValidationError,
)
from .models import CreateTaskRequest, Priority, TaskResult

logger = logging.getLogger(__name__)


def _timeout_seconds() -> float:
    raw_timeout = os.getenv("TASK_API_TIMEOUT_SECONDS", "10")
    try:
        timeout = float(raw_timeout)
    except ValueError as exc:
        raise RuntimeError("TASK_API_TIMEOUT_SECONDS must be a number.") from exc

    if timeout <= 0:
        raise RuntimeError("TASK_API_TIMEOUT_SECONDS must be greater than zero.")
    return timeout


mcp = FastMCP(name="task-manager", json_response=True)
client = TaskApiClient(
    base_url=os.getenv("TASK_API_BASE_URL", "http://localhost:8000"),
    timeout_seconds=_timeout_seconds(),
)


@mcp.tool()
async def create_task(
    title: Annotated[
        str,
        Field(
            min_length=1,
            max_length=100,
            description="Short task title.",
        ),
    ],
    description: Annotated[
        str | None,
        Field(
            max_length=1000,
            description="Optional details about the task.",
        ),
    ] = None,
    priority: Annotated[
        Priority,
        Field(description="Task priority: low, medium, or high."),
    ] = "medium",
    due_date: Annotated[
        date | None,
        Field(description="Optional due date in YYYY-MM-DD format."),
    ] = None,
) -> TaskResult:
    """Create one task in the task management service."""

    request = CreateTaskRequest(
        title=title,
        description=description,
        priority=priority,
        due_date=due_date,
    )

    try:
        return await client.create_task(request)
    except TaskApiValidationError as exc:
        raise ToolError(f"The task could not be created: {exc}") from exc
    except TaskApiUnavailableError as exc:
        raise ToolError(
            "The task service is temporarily unavailable. Try again later."
        ) from exc
    except TaskApiUnexpectedError as exc:
        logger.exception("Unexpected task API response")
        raise ToolError("The task service returned an unexpected response.") from exc


def main() -> None:
    """Run the MCP server over stdio without writing logs to stdout."""

    logging.basicConfig(
        level=logging.INFO,
        stream=sys.stderr,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

{
  "type": "object",
  "properties": {
    "title": {
      "type": "string",
      "minLength": 1,
      "maxLength": 100
    },
    "description": {
      "anyOf": [{"type": "string", "maxLength": 1000}, {"type": "null"}],
      "default": null
    },
    "priority": {
      "type": "string",
      "enum": ["low", "medium", "high"],
      "default": "medium"
    },
    "due_date": {
      "anyOf": [{"type": "string", "format": "date"}, {"type": "null"}],
      "default": null
    }
  },
  "required": ["title"]
}

The task could not be created: due_date must be today or later.

uv run task-api

uv run uvicorn task_api.app:app --reload --port 8000

TASK_API_BASE_URL=http://localhost:8000 \
npx -y @modelcontextprotocol/inspector \
uv run python -m task_mcp.server

codex mcp add task-manager \
  --env TASK_API_BASE_URL=http://localhost:8000 \
  --env TASK_API_TIMEOUT_SECONDS=10 \
  -- uv run python -m task_mcp.server

明日までに「MCPの記事を公開する」というタスクを作成してください。
優先度はhighにしてください。

{
  "title": "MCPの記事を公開する",
  "description": null,
  "priority": "high",
  "due_date": "2026-07-17"
}

cat data/tasks.json

uv run pytest

import pytest

from task_mcp import server


@pytest.mark.asyncio
async def test_create_task_schema_is_model_friendly() -> None:
    tools = await server.mcp.list_tools()
    create_task_tool = next(tool for tool in tools if tool.name == "create_task")
    schema = create_task_tool.inputSchema

    assert schema["required"] == ["title"]
    assert schema["properties"]["title"]["minLength"] == 1
    assert schema["properties"]["title"]["maxLength"] == 100
    assert schema["properties"]["priority"]["enum"] == [
        "low",
        "medium",
        "high",
    ]

POST   /tasks
GET    /tasks
GET    /tasks/{id}
PATCH  /tasks/{id}
DELETE /tasks/{id}

create_task
search_tasks
complete_task