restmcp 0.1.1

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/
├── datasource/        # external connections (APIs, databases)
├── models/            # domain entities (Pydantic)
├── repositories/      # data access layer
├── services/          # business logic
├── tools/             # internal utilities
├── urls/              # endpoint definitions (auto-discovery)
├── 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 datasource.product_api import ProductApiDataSource
from models.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 mcp_definition, url, method, and callback.

from restmcp import Endpoint
from services.product import GetProductService

class GetProductEndpoint(Endpoint):
    mcp_definition = {
        "name":        "get_product",
        "description": "Returns a product by ID",
        "parameters": {
            "properties": {
                "product_id": {"type": "string", "description": "Product ID"},
            },
        },
    }
    url    = "/api/get-product"
    method = "POST"

    async def callback(self, product_id: str) -> dict:
        return await GetProductService().execute(product_id)

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 and mcp_definition are missing

class GetUserEndpoint(BaseAuthEndpoint):
    mcp_definition = { ... }
    url = "/api/get-user"
# ↑ registered automatically: all required attributes 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()

Response format:

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

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

---

Server

Singleton with dual-mode: HTTP via FastAPI/uvicorn or MCP protocol via FastMCP.

from restmcp import Server
import urls  # triggers auto-discovery of all endpoint modules

server = Server.get_instance()

if __name__ == "__main__":
    server.start(host="0.0.0.0", port=5000)

# MCP mode
mcp = server.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)

---

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
    PythiaException --> ValidationError["ValidationError (400)"]
    PythiaException --> 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