EnrichMCP

The ORM for AI Agents - Turn your data model into a semantic MCP layer

EnrichMCP is a Python framework that helps AI agents understand and navigate your data. Built on MCP (Model Context Protocol), it adds a semantic layer that turns your data model into typed, discoverable tools - like an ORM for AI.

What is EnrichMCP?

Think of it as SQLAlchemy for AI agents. EnrichMCP automatically:

- Generates typed tools from your data models
- Handles relationships between entities (users → orders → products)
- Provides schema discovery so AI agents understand your data structure
- Validates all inputs/outputs with Pydantic models
- Works with any backend - databases, APIs, or custom logic

Installation

pip install enrichmcp
With SQLAlchemy support
pip install enrichmcp[sqlalchemy]

Show Me Code

Option 1: I Have SQLAlchemy Models (30 seconds)

Transform your existing SQLAlchemy models into an AI-navigable API:

from enrichmcp import EnrichMCP
from enrichmcp.sqlalchemy import (
    include_sqlalchemy_models,
    sqlalchemy_lifespan,
    EnrichSQLAlchemyMixin,
)
from sqlalchemy import ForeignKey
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
Add the mixin to your declarative base
class Base(DeclarativeBase, EnrichSQLAlchemyMixin):
    pass
class User(Base):

    """User account."""
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, info={"description": "Unique user ID"})

    email: Mapped[str] = mapped_column(unique=True, info={"description": "Email address"})

    status: Mapped[str] = mapped_column(default="active", info={"description": "Account status"})

    orders: Mapped[list["Order"]] = relationship(

        back_populates="user", info={"description": "All orders for this user"}

    )
class Order(Base):

    """Customer order."""
__tablename__ = "orders"
id: Mapped[int] = mapped_column(primary_key=True, info={"description": "Order ID"})

    user_id: Mapped[int] = mapped_column(

        ForeignKey("users.id"), info={"description": "Owner user ID"}

    )

    total: Mapped[float] = mapped_column(info={"description": "Order total"})

    user: Mapped[User] = relationship(

        back_populates="orders", info={"description": "User who placed the order"}

    )
That's it! Create your MCP app
app = EnrichMCP(
    "E-commerce Data",
    "API generated from SQLAlchemy models",
    lifespan=sqlalchemy_lifespan(Base, engine, cleanup_db_file=True),
)
include_sqlalchemy_models(app, Base)
if __name__ == "__main__":

    app.run()

Option 2: I Have REST APIs (2 minutes)

Wrap your existing APIs with semantic understanding:

from typing import Literal
from enrichmcp import EnrichMCP, EnrichModel, Relationship
from pydantic import Field
import httpx
app = EnrichMCP("API Gateway", "Wrapper around existing REST APIs")

http = httpx.AsyncClient(base_url="https://api.example.com")
@app.entity()

class Customer(EnrichModel):

    """Customer in our CRM system."""
id: int = Field(description="Unique customer ID")

    email: str = Field(description="Primary contact email")

    tier: Literal["free", "pro", "enterprise"] = Field(description="Subscription tier")
# Define navigable relationships

    orders: list["Order"] = Relationship(description="Customer's purchase history")
@app.entity()

class Order(EnrichModel):

    """Customer order from our e-commerce platform."""
id: int = Field(description="Order ID")

    customer_id: int = Field(description="Associated customer")

    total: float = Field(description="Order total in USD")

    status: Literal["pending", "shipped", "delivered"] = Field(description="Order status")
customer: Customer = Relationship(description="Customer who placed this order")
Define how to fetch data
@app.retrieve()
async def get_customer(customer_id: int) -> Customer:
    """Fetch customer from CRM API."""
    response = await http.get(f"/api/customers/{customer_id}")
    return Customer(response.json())
Define relationship resolvers
@Customer.orders.resolver
async def get_customer_orders(customer_id: int) -> list[Order]:
    """Fetch orders for a customer."""
    response = await http.get(f"/api/customers/{customer_id}/orders")
    return [Order(order) for order in response.json()]
@Order.customer.resolver

async def get_order_customer(order_id: int) -> Customer:

    """Fetch the customer for an order."""

    response = await http.get(f"/api/orders/{order_id}/customer")

    return Customer(response.json())

app.run()

@User.orders.resolver
async def list_user_orders(user_id: int) -> list[Order]:
"""Fetch orders for a user."""
rows = await db.query(
"SELECT FROM orders WHERE user_id = ? ORDER BY id DESC",
user_id,
)
return [Order(row) for row in rows]

@User.segments.resolver
async def list_user_segments(user_id: int) -> list[Segment]:
"""Fetch segments that include the user."""
rows = await db.query(
"SELECT s. FROM segments s JOIN user_segments us ON s.name = us.segment_name WHERE us.user_id = ?",
user_id,
)
return [Segment(**row) for row in rows]

…