Build an MCP Server for an Existing REST API: Let AI Manage Tasks with Python

Introduction: Why MCP When You Already Have a REST API?

REST APIs are excellent interfaces for software-to-software communication. If an API already exists, why not let an AI model call it directly?

You can provide an OpenAPI document to a model and ask it to construct HTTP requests. However, an AI client still needs a consistent way to discover available operations, understand their arguments, invoke them, and interpret their results.

An MCP Tool exposes a name, a description, and an input schema, and clients invoke it through tools/call. The MCP Tools specification defines tool metadata and JSON Schema-based inputs.

The central idea of this article is simple:

MCP does not replace a REST API. It adds an interface through which AI clients can use an existing API.

Instead of duplicating persistence and business rules inside the MCP server, we will build a thin AI-facing adapter around the existing REST API.

The complete runnable code for this article is available in the task-api-mcp GitHub repository. The file paths, commands, and tests below match that implementation.

1. System Overview

The user asks for a task in natural language. The AI client converts that request into structured arguments and calls the MCP server's create_task Tool over stdio. The MCP server translates the call into an HTTP request for the existing REST API.

flowchart LR
    accTitle: Overall task creation architecture
    accDescr: An AI client converts the user's natural-language request into an MCP Tool call, the MCP server sends an HTTP request to the REST API, and the API persists the task in a JSON file.
    User[User] -->|Natural language| AI[AI client]
    AI -->|MCP / stdio| MCP[MCP server]
    MCP -->|HTTP POST /tasks| API[Task REST API]
    API --> JSON[(data/tasks.json)]

End-to-end system data flow

Each component has a distinct responsibility.

A single user request crosses two communication boundaries: MCP between the AI client and adapter, then HTTP between the adapter and the existing system.

sequenceDiagram
    accTitle: Execution order for the create_task Tool
    accDescr: The AI client calls create_task, the MCP server validates the input and sends it to the REST API, and the API saves the task in tasks.json before returning the result along the same path.
    actor User
    participant AI as AI client
    participant MCP as MCP server
    participant API as REST API
    participant File as tasks.json
    User->>AI: Create a task due tomorrow
    AI->>MCP: tools/call create_task
    MCP->>MCP: Validate the input schema
    MCP->>API: POST /tasks
    API->>File: Save the task
    File-->>API: Save complete
    API-->>MCP: 201 Created
    MCP-->>AI: Structured creation result
    AI-->>User: The task was created

create_task execution sequence

The important design choice is that the MCP server does not become a second task-management system. The REST API remains the canonical entry point for task creation.

2. Review the Existing Task REST API

Assume the existing service provides these endpoints:

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

This article exposes only POST /tasks as an MCP Tool named create_task.

A request looks like this:

{
  "title": "Publish the MCP article",
  "description": "Review the diagrams and code",
  "priority": "high",
  "due_date": "2026-07-16"
}

On success, the API returns 201 Created and the persisted task.

{
  "id": "b48b8da3-f80b-40e3-a78d-8d70b11e58c4",
  "title": "Publish the MCP article",
  "description": "Review the diagrams and code",
  "priority": "high",
  "due_date": "2026-07-16",
  "status": "todo",
  "created_at": "2026-07-15T21:30:00+09:00"
}

The REST API validates required fields, text lengths, priority values, date syntax, and the rule that a due date cannot be in the past. Those checks must remain in place after MCP is introduced because other callers—web applications, scripts, and batch jobs—need the same guarantees.

Before adding MCP, verify the API independently:

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Publish the MCP article",
    "priority": "high",
    "due_date": "2026-07-16"
  }'

If this request fails, fix the REST API first. Debugging the adapter and the underlying service at the same time makes failures unnecessarily difficult to isolate.

3. Design the REST-to-MCP Adapter

Writing directly to data/tasks.json from create_task may look simpler, but it creates several problems:

• Persistence logic is duplicated across REST and MCP
• Validation and business rules have two sources of truth
• Authentication, rate limits, or audit logging may be bypassed
• A later database migration requires changes in both paths
• Behavior can differ depending on which interface a caller uses

Instead, the Tool calls TaskApiClient.create_task(), which sends POST /tasks to the existing service.

The Tool owns MCP inputs and outputs and translates known failures into meaningful Tool errors. The REST API retains the final decision about whether the task may be created.

If the REST API and MCP server live in the same codebase and process, both controllers could call a shared service layer. This article deliberately preserves the HTTP boundary because the goal is to adapt an independently deployed, existing service.

4. Prepare the Project from the Public Repository

Clone the completed example from the task-api-mcp repository. You need Python 3.10 or newer and uv. Node.js 18 or newer is only required for Inspector, and Codex CLI is only required for the Codex walkthrough.

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

The included uv.lock reproduces the dependency set tested for this article. As of July 16, 2026, 1.28.1 is the stable MCP Python SDK release and v2 remains a prerelease. The example pins mcp[cli]==1.28.1 so its v1 behavior stays reproducible. Check PyPI and the official v1.x repository when updating the project.

The package configuration also defines console commands for the REST API and MCP server.

[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",
]

The repository keeps the REST API and MCP adapter in separate packages.

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 owns validation and JSON persistence. task_mcp only translates between Tool calls and HTTP, and never modifies data/tasks.json directly.

The API location, timeout, and data path are configured through environment variables.

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

When using a local .env file, pass it explicitly to uv.

cp .env.example .env
uv run --env-file .env task-api
# In another terminal, when starting the MCP server directly
uv run --env-file .env task-mcp

5. Implement the REST API Client

Keep HTTP-specific behavior out of the Tool and translate REST outcomes into three application-level errors.

"""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."""

This prevents raw httpx exceptions and response bodies from leaking through the Tool while preserving enough meaning for the model to choose its next action.

"""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."

The client sets an explicit timeout, validates successful 201 responses, and separates correctable 4xx failures, temporary 5xx failures, and unexpected responses. It accepts an API detail only when it is a string and limits it to 300 characters instead of returning arbitrary internal structures.

6. Initialize the MCP Server

With v1.28.1, import FastMCP from the v1 package path:

from mcp.server.fastmcp import FastMCP

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

FastMCP derives Tool definitions from type hints and docstrings. Compatible return annotations, including Pydantic models, can also produce structured output schemas. The official v1.x README shows FastMCP initialization, @mcp.tool(), and Inspector usage.

The server name, Tool name, docstring, and argument descriptions serve different purposes:

• The server name identifies the configured connection
• The Tool name identifies the operation the model can invoke
• The docstring explains when and why to use it
• Argument descriptions explain the meaning and constraints of individual values

Prefer a specific name such as create_task over ambiguous names such as create or execute.

7. Implement the create_task Tool

Define structured models at the MCP boundary. The REST service has its own request model and remains responsible for final business-rule validation.

"""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

Keep Tool arguments flat and place both constraints and descriptions in the generated schema. The following is the complete server.py from the public repository.

"""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()

_timeout_seconds() validates configuration at startup and rejects non-numeric or non-positive values. main() sends logs to stderr so they cannot corrupt stdio traffic on stdout. The Tool itself only translates input, invokes the API client, and returns a structured result or a safe ToolError.

8. Design the Input Schema and Validation

The schema generated from the type hints and Field metadata will look approximately like this. Exact details may vary with SDK and Pydantic versions, so confirm the final output in Inspector.

{
  "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"]
}

Treat validation as two separate categories:

MCP-side validation provides faster feedback and avoids unnecessary HTTP calls. It must not replace validation in the REST service because MCP is not the only caller.

Errors should give the model enough information to choose its next action:

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

A generic Validation failed message does not identify which argument must change.

9. Return Structured Success and Safe Errors

On success, return TaskResult rather than only a sentence. The AI can explain the result to the user while retaining fields such as id, status, and due_date for later operations.

The MCP specification distinguishes two failure categories:

• Unknown Tools and malformed request structures are protocol errors
• API, input, and business-rule failures are Tool Execution Errors with isError: true

The latter should provide actionable feedback that allows a model to correct its request. See the error-handling section of the Tools specification. The v1 SDK defines ToolError for expected Tool failures; its definition is available in the v1.x source.

Never expose stack traces, local paths, API keys, internal hostnames, raw response bodies, or full library exceptions to the model. Record diagnostic details in server logs written to stderr.

10. Run over stdio and Inspect the Server

From the cloned repository root, start the REST API.

uv run task-api

The API listens on http://localhost:8000, and Swagger UI is available at http://localhost:8000/docs. For automatic reload during development, use:

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

Keep the API running and launch MCP Inspector from another terminal.

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

Use Inspector to verify:

• create_task appears in the Tool list
• Only title is required, with a 1–100 character constraint
• priority is a low, medium, or high enum
• due_date is represented as a date
• A valid task is created and persisted to data/tasks.json
• Past dates and a stopped API produce safe, actionable errors

With stdio, the client and server exchange JSON-RPC over stdin and stdout, so ordinary output must never be written to stdout. The repository uses logging.basicConfig(..., stream=sys.stderr) to keep diagnostics separate from protocol traffic. See the transport specification.

11. Invoke the Tool from an AI Client

From the repository root, register the stdio server with Codex CLI.

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

Verify it with codex mcp list, or use /mcp inside the Codex TUI. See the official Codex MCP documentation.

Then ask:

Create a task called "Publish the MCP article" due tomorrow.
Set its priority to high.

The model converts that request into Tool arguments such as:

{
  "title": "Publish the MCP article",
  "description": null,
  "priority": "high",
  "due_date": "2026-07-17"
}

due_date stores only a date. If time of day matters, add a datetime or time field to both the REST contract and Tool schema. The MCP adapter should not silently store information that the canonical API cannot represent.

Finally, inspect data/tasks.json to confirm that the natural-language request was persisted through the REST API.

cat data/tasks.json

12. Run the Automated Tests

The public repository tests the REST API, HTTP client, and MCP Tool as three separate layers.

uv run pytest

All 26 tests pass at commit 2b343d2. They use temporary directories and httpx.MockTransport, so no running API is required and your local data/tasks.json is not modified.

For example, the MCP tests inspect the schema that the server actually publishes.

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",
    ]

The three test layers cover:

1. REST API: successful creation, whitespace trimming, unknown fields, past dates, 404s, and storage failures
2. API client: 201, 422, timeouts, connection failures, 5xx, invalid JSON, and unknown HTTP states
3. MCP Tool: structured results, schema constraints, validation failures, temporary outages, and safe ToolError conversion

See the GitHub repository for the full tests and current implementation.

13. Prepare for Production

Keep credentials out of Tool arguments

Production adapters may forward API keys, Bearer tokens, user IDs, tenant IDs, or trace IDs. Do not expose credentials as arguments for the model to generate. Read them from the MCP server's environment, a secret store, or an authenticated session.

Make creation idempotent

AI clients and network layers may retry operations. Consider Idempotency-Key, request IDs, duplicate detection, execution history, and audit logs in the REST service. Preventing duplicates only inside the MCP adapter does not protect other API clients.

Choose between stdio and Streamable HTTP

stdio is a good fit when a local AI client launches a local process. A remote, multi-user deployment requires Streamable HTTP plus authentication, session management, Origin validation, and appropriate network controls. That remote architecture is outside this article's scope.

Do not convert every REST endpoint mechanically

Even if the REST API exposes these routes, the MCP server does not need a one-to-one Tool mapping:

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

Design Tools around safe, understandable user intents instead:

create_task
search_tasks
complete_task

REST APIs represent resources and system boundaries. MCP Tools represent operations that a model can understand and invoke safely. High-risk actions such as deletion or external publication particularly benefit from separate Tools, clear descriptions, approval requirements, and audit trails.

Conclusion

Adding MCP support to an existing REST service does not require removing the API or reimplementing persistence for AI callers.

The MCP server can remain a thin AI-facing adapter: translate Tool inputs into an API request, return structured results, and convert API failures into safe, actionable errors. Keep business rules and persistence behind the canonical REST boundary.

This separation preserves existing API clients while adding natural language as a new interaction path. After validating the local stdio setup, the design can grow incrementally to include authentication, idempotency, audit logging, and Streamable HTTP.

Start by cloning the public example, then run the REST API, Inspector, Codex, and pytest to observe each boundary directly.