enrichmcp

# EnrichMCP **The ORM for AI Agents - Turn your data model into a semantic MCP layer** [](https://github.com/featureform/enrichmcp/actions/workflows/ci.yml) [](https://codecov.io/gh/featureform/enrichmcp) [](https://pypi.org/project/enrichmcp/) [](https://www.python.org/downloads/) [](https://github.com/featureform/enrichmcp/blob/main/LICENSE) [](https://featureform.github.io/enrichmcp) 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 ```bash 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: ```python 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: ```python 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