Skip to main content
Glama
featureform

EnrichMCP

Official
by featureform
OverviewSchemaRelated ServersScoreDiscussions
Python
Hybrid

EnrichMCP

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

CI Coverage PyPI Python 3.11+ License Docs

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

AI agents can now:

  • explore_data_model() - understand your entire schema

  • list_users(status='active') - query with filters

  • get_user(id=123) - fetch specific records

  • Navigate relationships: user.orders → order.user

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

Option 3: I Want Full Control (5 minutes)

Build a complete data layer with custom logic:

from enrichmcp import EnrichMCP, EnrichModel, Relationship
from datetime import datetime
from decimal import Decimal
from pydantic import Field

app = EnrichMCP("Analytics Platform", "Custom analytics API")

db = ...  # your database connection


@app.entity()
class User(EnrichModel):
    """User with computed analytics fields."""

    id: int = Field(description="User ID")
    email: str = Field(description="Contact email")
    created_at: datetime = Field(description="Registration date")

    # Computed fields
    lifetime_value: Decimal = Field(description="Total revenue from user")
    churn_risk: float = Field(description="ML-predicted churn probability 0-1")

    # Relationships
    orders: list["Order"] = Relationship(description="Purchase history")
    segments: list["Segment"] = Relationship(description="Marketing segments")


@app.entity()
class Segment(EnrichModel):
    """Dynamic user segment for marketing."""

    name: str = Field(description="Segment name")
    criteria: dict = Field(description="Segment criteria")
    users: list[User] = Relationship(description="Users in this segment")


@app.entity()
class Order(EnrichModel):
    """Simplified order record."""

    id: int = Field(description="Order ID")
    user_id: int = Field(description="Owner user ID")
    total: Decimal = Field(description="Order total")


@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]


@Segment.users.resolver
async def list_segment_users(name: str) -> list[User]:
    """List users in a segment."""
    rows = await db.query(
        "SELECT u.* FROM users u JOIN user_segments us ON u.id = us.user_id WHERE us.segment_name = ?",
        name,
    )
    return [User(**row) for row in rows]


# Complex resource with business logic
@app.retrieve()
async def find_high_value_at_risk_users(
    lifetime_value_min: Decimal = 1000, churn_risk_min: float = 0.7, limit: int = 100
) -> list[User]:
    """Find valuable customers likely to churn."""
    users = await db.query(
        """
        SELECT * FROM users
        WHERE lifetime_value >= ? AND churn_risk >= ?
        ORDER BY lifetime_value DESC
        LIMIT ?
        """,
        lifetime_value_min,
        churn_risk_min,
        limit,
    )
    return [User(**u) for u in users]


# Async computed field resolver
@User.lifetime_value.resolver
async def calculate_lifetime_value(user_id: int) -> Decimal:
    """Calculate total revenue from user's orders."""
    total = await db.query_single("SELECT SUM(total) FROM orders WHERE user_id = ?", user_id)
    return Decimal(str(total or 0))


# ML-powered field
@User.churn_risk.resolver
async def predict_churn_risk(user_id: int) -> float:
    """Run churn prediction model."""
    ctx = app.get_context()
    features = await gather_user_features(user_id)
    model = ctx.get("ml_models")["churn"]
    return float(model.predict_proba(features)[0][1])


app.run()

Key Features

🔍 Automatic Schema Discovery

AI agents explore your entire data model with one call:

schema = await explore_data_model()
# Returns complete schema with entities, fields, types, and relationships

🔗 Relationship Navigation

Define relationships once, AI agents traverse naturally:

# AI can navigate: user → orders → products → categories
user = await get_user(123)
orders = await user.orders()  # Automatic resolver
products = await orders[0].products()

🛡️ Type Safety & Validation

Full Pydantic validation on every interaction:

@app.entity()
class Order(EnrichModel):
    total: float = Field(ge=0, description="Must be positive")
    email: EmailStr = Field(description="Customer email")
    status: Literal["pending", "shipped", "delivered"]

describe_model() will list these allowed values so agents know the valid options.

✏️ Mutability & CRUD

Fields are immutable by default. Mark them as mutable and use auto-generated patch models for updates:

@app.entity()
class Customer(EnrichModel):
    id: int = Field(description="ID")
    email: str = Field(json_schema_extra={"mutable": True}, description="Email")


@app.create()
async def create_customer(email: str) -> Customer: ...


@app.update()
async def update_customer(cid: int, patch: Customer.PatchModel) -> Customer: ...


@app.delete()
async def delete_customer(cid: int) -> bool: ...

📄 Pagination Built-in

Handle large datasets elegantly:

from enrichmcp import PageResult


@app.retrieve()
async def list_orders(page: int = 1, page_size: int = 50) -> PageResult[Order]:
    orders, total = await db.get_orders_page(page, page_size)
    return PageResult.create(items=orders, page=page, page_size=page_size, total_items=total)

See the Pagination Guide for more examples.

🔐 Context & Authentication

Pass auth, database connections, or any context:

from pydantic import Field
from enrichmcp import EnrichModel


class UserProfile(EnrichModel):
    """User profile information."""

    user_id: int = Field(description="User ID")
    bio: str | None = Field(default=None, description="Short bio")


@app.retrieve()
async def get_user_profile(user_id: int) -> UserProfile:
    ctx = app.get_context()
    # Access context provided by MCP client
    auth_user = ctx.get("authenticated_user_id")
    if auth_user != user_id:
        raise PermissionError("Can only access your own profile")
    return await db.get_profile(user_id)

⚡ Request Caching

Reduce API overhead by storing results in a per-request, per-user, or global cache:

@app.retrieve()
async def get_customer(cid: int) -> Customer:
    ctx = app.get_context()

    async def fetch() -> Customer:
        return await db.get_customer(cid)

    return await ctx.cache.get_or_set(f"customer:{cid}", fetch)

🧭 Parameter Hints

Provide examples and metadata for tool parameters using EnrichParameter:

from enrichmcp import EnrichParameter


@app.retrieve()
async def greet_user(name: str = EnrichParameter(description="user name", examples=["bob"])) -> str:
    return f"Hello {name}"

Tool descriptions will include the parameter type, description, and examples.

🌐 HTTP & SSE Support

Serve your API over standard output (default), SSE, or HTTP:

app.run()  # stdio default
app.run(transport="streamable-http")

Why EnrichMCP?

EnrichMCP adds three critical layers on top of MCP:

  1. Semantic Layer - AI agents understand what your data means, not just its structure

  2. Data Layer - Type-safe models with validation and relationships

  3. Control Layer - Authentication, pagination, and business logic

The result: AI agents can work with your data as naturally as a developer using an ORM.

Server-Side LLM Sampling

EnrichMCP can request language model completions through MCP's sampling feature. Call ctx.ask_llm() or the ctx.sampling() alias from any resource and the connected client will choose an LLM and pay for the usage. You can tune behavior using options like model_preferences, allow_tools, and max_tokens. See docs/server_side_llm.md for more details.

Examples

Check out the examples directory:

Documentation

Contributing

We welcome contributions! See CONTRIBUTING.md for details.

Development Setup

The repository requires Python 3.11 or newer. The Makefile includes commands to create a virtual environment and run the tests:

make setup            # create .venv and install dependencies
source .venv/bin/activate
make test             # run the test suite

This installs all development extras and pre-commit hooks so commands like make lint or make docs work right away.

License

Apache 2.0 - See LICENSE

Built by Featureform • MCP Protocol

This server cannot be installed

A
license - permissive license
-
quality - not tested
D
maintenance

How are these scores calculated?

Maintenance

–Maintainers
–Response time
4dRelease cycle
8Releases (12mo)
Issues opened vs closed

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/featureform/enrichmcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server