restmcp 0.1.5

Python framework for building MCP servers with a layered architecture and REST compatibility

restmcp

One framework. MCP tools and REST endpoints, auto-registered.

Python framework for building MCP servers with a layered architecture and REST compatibility.
Annotated classes become MCP tools and HTTP endpoints: auto-registered, dependency-injected, sync/async agnostic.

---

Architecture

graph LR
    LLM["🤖 LLM / Client"] -->|"HTTP or MCP"| EP["Endpoint"]
    EP --> SV["Service"]
    SV --> RP["Repository"]
    RP --> DS["DataSource"]
    DS --> EX[("External\nAPI / DB")]

    style EP fill:#4f46e5,color:#fff,stroke:none
    style SV fill:#7c3aed,color:#fff,stroke:none
    style RP fill:#9333ea,color:#fff,stroke:none
    style DS fill:#a855f7,color:#fff,stroke:none

Each layer knows only the layer directly below it. Every class name is suffix-enforced at import time: a typo raises TypeError before the server starts.

---

Installation

pip install restmcp

---

Quick start

restmcp new my-server
cd my-server
pip install -e .
python main.py

Generated structure:

my-server/
├── datasources/       # external connections (APIs, databases)
├── entities/          # domain models (Pydantic)
├── repositories/      # data access layer
├── services/          # business logic
├── utils/             # internal helpers
├── endpoints/         # endpoint definitions (auto-discovered)
├── main.py
└── pyproject.toml

---

How it works

sequenceDiagram
    participant C as Client / LLM
    participant E as Endpoint
    participant S as Service
    participant R as Repository
    participant D as DataSource

    C->>E: POST /api/get-product {"product_id": "1"}
    E->>S: service.execute(product_id="1")
    S->>R: repo.get(product_id="1")
    R->>D: data_source.fetch("1")
    D-->>R: raw dict
    R-->>S: ProductEntity
    S-->>E: result dict
    E-->>C: {"tool": "get_product", "result": {...}, "success": true}

---

Base classes

DataSource

Abstracts the connection to an external data source (REST API, database, file).
Rule: class name must end with DataSource.

import httpx
from restmcp import DataSource

class ProductApiDataSource(DataSource):
    base_url = "https://api.example.com"

    async def fetch(self, product_id: str) -> dict:
        async with httpx.AsyncClient() as client:
            r = await client.get(f"{self.base_url}/products/{product_id}")
            r.raise_for_status()
            return r.json()

---

Entity

Structured domain data backed by Pydantic. Automatic type validation.
Rule: class name must end with Entity.

from restmcp import Entity

class ProductEntity(Entity):
    id:    str
    name:  str
    price: float

---

Repository

Fetches data via a DataSource and returns Entity objects. One source, one data type.
Rules: name ends with Repository; must declare data_source as class attribute; must implement get().

from restmcp import Repository
from datasources.product_api import ProductApiDataSource
from entities.product import ProductEntity

class ProductRepository(Repository):
    data_source = ProductApiDataSource()

    async def get(self, product_id: str) -> ProductEntity:
        raw = await self.data_source.fetch(product_id)
        return ProductEntity(**raw)

Dependency injection:

repo = ProductRepository()                              # uses real DataSource
repo = ProductRepository(data_source=MockDataSource())  # injects mock for tests

Repository.__init__ uses copy.copy() of the class attribute: instances are always isolated.

---

Service

Orchestrates business logic. Where joins, transformations, and multi-source rules live.
Rules: name ends with Service; must declare at least one Repository as class attribute.

from restmcp import Service
from repositories.product import ProductRepository

class GetProductService(Service):
    repo = ProductRepository()

    async def execute(self, product_id: str) -> dict:
        product = await self.repo.get(product_id=product_id)
        return product.model_dump()

Dependency injection:

svc = GetProductService()                       # production
svc = GetProductService(repo=MockRepository())  # test

Repository class attributes are auto-discovered via MRO and isolated per instance.

---

Endpoint

HTTP + MCP route. Auto-registers on class definition: no manual wiring needed.
Rules: name ends with Endpoint; must declare url, method, and callback. mcp_definition is inferred from the callback when omitted.

from typing import Annotated

from restmcp import Endpoint
from services.product import GetProductService

class GetProductEndpoint(Endpoint):
    url    = "/api/get-product"
    method = "POST"

    async def callback(self, product_id: Annotated[str, "Product ID"]) -> dict:
        """Returns a product by ID."""
        return await GetProductService().execute(product_id)

The MCP tool name (get_product), description (the docstring's first line), and parameter schema (types + Annotated text) are inferred from the callback. Set mcp_definition explicitly only when you need to override the inferred schema.

Defining the class is enough. The route is registered on the Server singleton the moment Python processes the class body.

Disabling an endpoint:

class GetProductEndpoint(Endpoint):
    disabled = True  # skips auto-registration; can still be instantiated manually
    ...

Abstract base classes (missing any required attribute) are never auto-registered:

class BaseAuthEndpoint(Endpoint):
    method = "POST"
    def callback(self, **kwargs): ...
# ↑ not registered: url is missing

class GetUserEndpoint(BaseAuthEndpoint):
    url = "/api/get-user"
    def callback(self, user_id: str) -> dict: ...
# ↑ registered automatically: url, method, and callback all present

Sync and async callbacks are both supported: restmcp detects and handles either:

# sync: runs in a thread pool, does not block the event loop
def callback(self, product_id: str) -> dict:
    return requests.get(f"https://api.example.com/products/{product_id}").json()

# async: awaited directly; use asyncio.gather for parallel I/O
async def callback(self, product_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.example.com/products/{product_id}")
        return r.json()

The contract (identical for REST and MCP):

• A sync callback runs in a threadpool, so blocking work — a synchronous DB driver, requests, file I/O — never stalls the event loop. Writing your Repository/DataSource synchronously is the simple, correct default.
• An async callback is awaited directly. Inside it, keep the I/O async (httpx, an async DB driver): calling blocking code from an async callback does stall the loop, because it is not moved to a thread.

Rule of thumb: sync all the way down, or async all the way down — don't put blocking calls inside an async def callback.

Response format:

{ "tool": "get_product", "result": { ... }, "success": true }

{ "tool": "get_product", "error": "not found", "error_type": "NotFoundError", "success": false }

---

Server

Singleton serving REST (FastAPI/uvicorn) and the MCP protocol (FastMCP) from one codebase. The recommended entry point is asgi_app(), which mounts both:

import uvicorn

from restmcp import Server, autodiscover

autodiscover("endpoints")  # imports every endpoint module so each one registers

app = Server.get_instance().asgi_app()  # REST at "/", MCP at "/mcp-protocol/"

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# REST only (no MCP), if that is all you need:
Server.get_instance().start(host="0.0.0.0", port=5000)

# The raw FastMCP instance (escape hatch):
mcp = Server.get_instance().get_mcp()

Built-in routes:

Route	Method	Auth required
/health	GET	No
/mcp/tools	GET	No
your endpoints	POST	Yes (if AUTH_API_KEY is set)

---

Production checklist

• Running REST + MCP together: use app = server.asgi_app() — it mounts both and wires the FastMCP lifespan. Do not call server.app.mount(...) directly: it raises "Task group is not initialized" on the first MCP request. MCP is served at the mcp_path you pass (default /mcp-protocol/, trailing slash); REST stays at /.
• Auth: set AUTH_API_KEY (Bearer). asgi_app() protects REST and the mounted MCP; /health and /mcp/tools remain public. Multiple keys: comma-separated.
• Serialization: callback return values are serialized with jsonable_encoder — datetime → ISO 8601, Decimal → string, Pydantic models → dict, automatically. Override per-entity via serialize().
• Typed parameters: a parameter declared as {"type": "string", "format": "date-time"} arrives in the callback as a string — coerce to datetime if needed.
• Caching: wrap an expensive Service method with @cached_method(ttl=seconds, maxsize=128) — the cache key is built from the arguments (via repr), so it works even with list/dict args. The store is bounded (maxsize) and self-purging (TTL), so it is safe in long-running processes. Cache plain-data arguments, not rich objects.
• Folders vs suffixes: only class suffixes are enforced (*Entity, *Repository, *Service, *Endpoint, *DataSource); folder names are free.
• Dependencies: fastmcp 3.x is recommended (this package requires fastmcp>=2.0; upgrade to 3.x for full Streamable HTTP support). Installing fastmcp also pulls in starlette.

A complete, runnable server using all of the above lives in examples/telemetry/ — no database required.

---

Exceptions

Raised inside callback: caught by Endpoint and converted to HTTP responses automatically.

from restmcp import ValidationError, NotFoundError

raise ValidationError("product_id is required")  # → HTTP 400
raise NotFoundError("Product not found")          # → HTTP 404

graph TD
    RestMCPException --> ValidationError["ValidationError (400)"]
    RestMCPException --> NotFoundError["NotFoundError (404)"]

---

Testing with injection

from restmcp import DataSource
from repositories.product import ProductRepository
from services.product import GetProductService

class FakeProductApiDataSource(DataSource):
    async def fetch(self, product_id: str) -> dict:
        return {"id": product_id, "name": "Test Widget", "price": 1.99}

def test_get_product():
    svc = GetProductService(repo=ProductRepository(data_source=FakeProductApiDataSource()))
    result = svc.execute(product_id="1")
    assert result["name"] == "Test Widget"

---

Environment variables

Variable	Default	Description
AUTH_API_KEY	(disabled)	Bearer token. Multiple keys supported comma-separated.
CORS_ORIGINS	*	Allowed origins. Multiple values supported comma-separated.
LOG_LEVEL	INFO	Log level: DEBUG, INFO, WARNING, ERROR.

---

Naming conventions

All base classes enforce a suffix. Violating it raises TypeError at import time: before the server starts.

Base class	Required suffix	Example
DataSource	*DataSource	ProductApiDataSource
Entity	*Entity	ProductEntity
Repository	*Repository	ProductRepository
Service	*Service	GetProductService
Endpoint	*Endpoint	GetProductEndpoint

---

Dependencies

fastapi    >= 0.100
uvicorn    >= 0.20
fastmcp    >= 2.0
pydantic   >= 2.0
click      >= 8.0

---

Author

Jorge Henrique Moreira Santana
Electrical Engineer, Postgraduate in Artificial Intelligence
LinkedIn · [e-mail]

---

License

MIT