# 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) -> 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:
```python
from enrichmcp import EnrichMCP, EnrichModel, Relationship, EnrichContext
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, context: EnrichContext) -> float:
"""Run churn prediction model."""
features = await gather_user_features(user_id)
model = context.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:
```python
schema = await explore_data_model()
# Returns complete schema with entities, fields, types, and relationships
```
### 🔗 Relationship Navigation
Define relationships once, AI agents traverse naturally:
```python
# 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:
```python
@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:
```python
@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:
```python
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](https://featureform.github.io/enrichmcp/pagination) for more examples.
### 🔐 Context & Authentication
Pass auth, database connections, or any context:
```python
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, context: EnrichContext) -> UserProfile:
# Access context provided by MCP client
auth_user = context.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:
```python
@app.retrieve
async def get_customer(cid: int, ctx: EnrichContext) -> Customer:
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`:
```python
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:
```python
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.
## Examples
Check out the [examples directory](examples/README.md):
- [hello_world](examples/hello_world) - The smallest possible EnrichMCP app
- [hello_world_http](examples/hello_world_http) - HTTP example using streamable HTTP
- [shop_api](examples/shop_api) - In-memory shop API with pagination and filters
- [shop_api_sqlite](examples/shop_api_sqlite) - SQLite-backed version
- [shop_api_gateway](examples/shop_api_gateway) - EnrichMCP as a gateway in front of FastAPI
- [sqlalchemy_shop](examples/sqlalchemy_shop) - Auto-generated API from SQLAlchemy models
- [mutable_crud](examples/mutable_crud) - Demonstrates mutable fields and CRUD decorators
- [caching](examples/caching) - Demonstrates ContextCache usage
- [basic_memory](examples/basic_memory) - Simple note-taking API using FileMemoryStore
- [openai_chat_agent](examples/openai_chat_agent) - Interactive chat client for MCP examples
## Documentation
- 📖 [Full Documentation](https://featureform.github.io/enrichmcp)
- 🚀 [Getting Started Guide](https://featureform.github.io/enrichmcp/getting-started)
- 🔧 [API Reference](https://featureform.github.io/enrichmcp/api)
## Contributing
We welcome contributions! See [CONTRIBUTING.md](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:
```bash
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](LICENSE)
---
Built by [Featureform](https://featureform.com) • [MCP Protocol](https://modelcontextprotocol.io)
Raw data
{
"_id": null,
"home_page": null,
"name": "enrichmcp",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": "Featureform Team <team@featureform.com>",
"keywords": "mcp, ai, agents, llm, model-context-protocol, pydantic, api, data-model, orm, agentic, enrichment",
"author": null,
"author_email": "Simba Khadder <simba@featureform.com>",
"download_url": "https://files.pythonhosted.org/packages/ee/1f/681ed55003f8ce2064931ff05d6c5c2e85dcda6d4dda3dcb34d92c14a9b8/enrichmcp-0.4.6.tar.gz",
"platform": null,
"description": "# EnrichMCP\n\n**The ORM for AI Agents - Turn your data model into a semantic MCP layer**\n\n[](https://github.com/featureform/enrichmcp/actions/workflows/ci.yml)\n[](https://codecov.io/gh/featureform/enrichmcp)\n[](https://pypi.org/project/enrichmcp/)\n[](https://www.python.org/downloads/)\n[](https://github.com/featureform/enrichmcp/blob/main/LICENSE)\n[](https://featureform.github.io/enrichmcp)\n\nEnrichMCP 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.\n\n## What is EnrichMCP?\n\nThink of it as SQLAlchemy for AI agents. EnrichMCP automatically:\n\n- **Generates typed tools** from your data models\n- **Handles relationships** between entities (users \u2192 orders \u2192 products)\n- **Provides schema discovery** so AI agents understand your data structure\n- **Validates all inputs/outputs** with Pydantic models\n- **Works with any backend** - databases, APIs, or custom logic\n\n## Installation\n\n```bash\npip install enrichmcp\n\n# With SQLAlchemy support\npip install enrichmcp[sqlalchemy]\n```\n\n## Show Me Code\n\n### Option 1: I Have SQLAlchemy Models (30 seconds)\n\nTransform your existing SQLAlchemy models into an AI-navigable API:\n\n\n```python\nfrom enrichmcp import EnrichMCP\nfrom enrichmcp.sqlalchemy import include_sqlalchemy_models, sqlalchemy_lifespan, EnrichSQLAlchemyMixin\nfrom sqlalchemy import ForeignKey\nfrom sqlalchemy.ext.asyncio import create_async_engine\nfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship\n\nengine = create_async_engine(\"postgresql+asyncpg://user:pass@localhost/db\")\n\n# Add the mixin to your declarative base\nclass Base(DeclarativeBase, EnrichSQLAlchemyMixin):\n pass\n\nclass User(Base):\n \"\"\"User account.\"\"\"\n\n __tablename__ = \"users\"\n\n id: Mapped[int] = mapped_column(primary_key=True, info={\"description\": \"Unique user ID\"})\n email: Mapped[str] = mapped_column(unique=True, info={\"description\": \"Email address\"})\n status: Mapped[str] = mapped_column(default=\"active\", info={\"description\": \"Account status\"})\n orders: Mapped[list[\"Order\"]] = relationship(back_populates=\"user\", info={\"description\": \"All orders for this user\"})\n\nclass Order(Base):\n \"\"\"Customer order.\"\"\"\n\n __tablename__ = \"orders\"\n\n id: Mapped[int] = mapped_column(primary_key=True, info={\"description\": \"Order ID\"})\n user_id: Mapped[int] = mapped_column(ForeignKey(\"users.id\"), info={\"description\": \"Owner user ID\"})\n total: Mapped[float] = mapped_column(info={\"description\": \"Order total\"})\n user: Mapped[User] = relationship(back_populates=\"orders\", info={\"description\": \"User who placed the order\"})\n\n# That's it! Create your MCP app\napp = EnrichMCP(\n \"E-commerce Data\",\n \"API generated from SQLAlchemy models\",\n lifespan=sqlalchemy_lifespan(Base, engine, cleanup_db_file=True),\n)\ninclude_sqlalchemy_models(app, Base)\n\nif __name__ == \"__main__\":\n app.run()\n```\nAI agents can now:\n- `explore_data_model()` - understand your entire schema\n- `list_users(status='active')` - query with filters\n- `get_user(id=123)` - fetch specific records\n- Navigate relationships: `user.orders` \u2192 `order.user`\n\n### Option 2: I Have REST APIs (2 minutes)\n\nWrap your existing APIs with semantic understanding:\n\n```python\nfrom typing import Literal\nfrom enrichmcp import EnrichMCP, EnrichModel, Relationship\nfrom pydantic import Field\nimport httpx\n\napp = EnrichMCP(\"API Gateway\", \"Wrapper around existing REST APIs\")\nhttp = httpx.AsyncClient(base_url=\"https://api.example.com\")\n\n@app.entity\nclass Customer(EnrichModel):\n \"\"\"Customer in our CRM system.\"\"\"\n\n id: int = Field(description=\"Unique customer ID\")\n email: str = Field(description=\"Primary contact email\")\n tier: Literal[\"free\", \"pro\", \"enterprise\"] = Field(\n description=\"Subscription tier\"\n )\n\n # Define navigable relationships\n orders: list[\"Order\"] = Relationship(description=\"Customer's purchase history\")\n\n@app.entity\nclass Order(EnrichModel):\n \"\"\"Customer order from our e-commerce platform.\"\"\"\n\n id: int = Field(description=\"Order ID\")\n customer_id: int = Field(description=\"Associated customer\")\n total: float = Field(description=\"Order total in USD\")\n status: Literal[\"pending\", \"shipped\", \"delivered\"] = Field(\n description=\"Order status\"\n )\n\n customer: Customer = Relationship(description=\"Customer who placed this order\")\n\n# Define how to fetch data\n@app.retrieve\nasync def get_customer(customer_id: int) -> Customer:\n \"\"\"Fetch customer from CRM API.\"\"\"\n response = await http.get(f\"/api/customers/{customer_id}\")\n return Customer(**response.json())\n\n# Define relationship resolvers\n@Customer.orders.resolver\nasync def get_customer_orders(customer_id: int) -> list[Order]:\n \"\"\"Fetch orders for a customer.\"\"\"\n response = await http.get(f\"/api/customers/{customer_id}/orders\")\n return [Order(**order) for order in response.json()]\n\n@Order.customer.resolver\nasync def get_order_customer(order_id: int) -> Customer:\n \"\"\"Fetch the customer for an order.\"\"\"\n response = await http.get(f\"/api/orders/{order_id}/customer\")\n return Customer(**response.json())\n\napp.run()\n```\n\n### Option 3: I Want Full Control (5 minutes)\n\nBuild a complete data layer with custom logic:\n\n```python\nfrom enrichmcp import EnrichMCP, EnrichModel, Relationship, EnrichContext\nfrom datetime import datetime\nfrom decimal import Decimal\nfrom pydantic import Field\n\napp = EnrichMCP(\"Analytics Platform\", \"Custom analytics API\")\n\ndb = ... # your database connection\n\n@app.entity\nclass User(EnrichModel):\n \"\"\"User with computed analytics fields.\"\"\"\n\n id: int = Field(description=\"User ID\")\n email: str = Field(description=\"Contact email\")\n created_at: datetime = Field(description=\"Registration date\")\n\n # Computed fields\n lifetime_value: Decimal = Field(description=\"Total revenue from user\")\n churn_risk: float = Field(description=\"ML-predicted churn probability 0-1\")\n\n # Relationships\n orders: list[\"Order\"] = Relationship(description=\"Purchase history\")\n segments: list[\"Segment\"] = Relationship(description=\"Marketing segments\")\n\n@app.entity\nclass Segment(EnrichModel):\n \"\"\"Dynamic user segment for marketing.\"\"\"\n\n name: str = Field(description=\"Segment name\")\n criteria: dict = Field(description=\"Segment criteria\")\n users: list[User] = Relationship(description=\"Users in this segment\")\n\n\n@app.entity\nclass Order(EnrichModel):\n \"\"\"Simplified order record.\"\"\"\n\n id: int = Field(description=\"Order ID\")\n user_id: int = Field(description=\"Owner user ID\")\n total: Decimal = Field(description=\"Order total\")\n\n@User.orders.resolver\nasync def list_user_orders(user_id: int) -> list[Order]:\n \"\"\"Fetch orders for a user.\"\"\"\n rows = await db.query(\n \"SELECT * FROM orders WHERE user_id = ? ORDER BY id DESC\",\n user_id,\n )\n return [Order(**row) for row in rows]\n\n@User.segments.resolver\nasync def list_user_segments(user_id: int) -> list[Segment]:\n \"\"\"Fetch segments that include the user.\"\"\"\n rows = await db.query(\n \"SELECT s.* FROM segments s JOIN user_segments us ON s.name = us.segment_name WHERE us.user_id = ?\",\n user_id,\n )\n return [Segment(**row) for row in rows]\n\n@Segment.users.resolver\nasync def list_segment_users(name: str) -> list[User]:\n \"\"\"List users in a segment.\"\"\"\n rows = await db.query(\n \"SELECT u.* FROM users u JOIN user_segments us ON u.id = us.user_id WHERE us.segment_name = ?\",\n name,\n )\n return [User(**row) for row in rows]\n\n# Complex resource with business logic\n@app.retrieve\nasync def find_high_value_at_risk_users(\n lifetime_value_min: Decimal = 1000,\n churn_risk_min: float = 0.7,\n limit: int = 100\n) -> list[User]:\n \"\"\"Find valuable customers likely to churn.\"\"\"\n users = await db.query(\n \"\"\"\n SELECT * FROM users\n WHERE lifetime_value >= ? AND churn_risk >= ?\n ORDER BY lifetime_value DESC\n LIMIT ?\n \"\"\",\n lifetime_value_min, churn_risk_min, limit\n )\n return [User(**u) for u in users]\n\n# Async computed field resolver\n@User.lifetime_value.resolver\nasync def calculate_lifetime_value(user_id: int) -> Decimal:\n \"\"\"Calculate total revenue from user's orders.\"\"\"\n total = await db.query_single(\n \"SELECT SUM(total) FROM orders WHERE user_id = ?\",\n user_id\n )\n return Decimal(str(total or 0))\n\n# ML-powered field\n@User.churn_risk.resolver\nasync def predict_churn_risk(user_id: int, context: EnrichContext) -> float:\n \"\"\"Run churn prediction model.\"\"\"\n features = await gather_user_features(user_id)\n model = context.get(\"ml_models\")[\"churn\"]\n return float(model.predict_proba(features)[0][1])\n\napp.run()\n```\n\n## Key Features\n\n### \ud83d\udd0d Automatic Schema Discovery\n\nAI agents explore your entire data model with one call:\n\n```python\nschema = await explore_data_model()\n# Returns complete schema with entities, fields, types, and relationships\n```\n\n### \ud83d\udd17 Relationship Navigation\n\nDefine relationships once, AI agents traverse naturally:\n\n```python\n# AI can navigate: user \u2192 orders \u2192 products \u2192 categories\nuser = await get_user(123)\norders = await user.orders() # Automatic resolver\nproducts = await orders[0].products()\n```\n\n### \ud83d\udee1\ufe0f Type Safety & Validation\n\nFull Pydantic validation on every interaction:\n\n```python\n@app.entity\nclass Order(EnrichModel):\n total: float = Field(ge=0, description=\"Must be positive\")\n email: EmailStr = Field(description=\"Customer email\")\n status: Literal[\"pending\", \"shipped\", \"delivered\"]\n```\n`describe_model()` will list these allowed values so agents know the valid options.\n\n### \u270f\ufe0f Mutability & CRUD\n\nFields are immutable by default. Mark them as mutable and use\nauto-generated patch models for updates:\n\n```python\n@app.entity\nclass Customer(EnrichModel):\n id: int = Field(description=\"ID\")\n email: str = Field(json_schema_extra={\"mutable\": True}, description=\"Email\")\n\n@app.create\nasync def create_customer(email: str) -> Customer:\n ...\n\n@app.update\nasync def update_customer(cid: int, patch: Customer.PatchModel) -> Customer:\n ...\n\n@app.delete\nasync def delete_customer(cid: int) -> bool:\n ...\n```\n\n### \ud83d\udcc4 Pagination Built-in\n\nHandle large datasets elegantly:\n\n```python\nfrom enrichmcp import PageResult\n\n@app.retrieve\nasync def list_orders(\n page: int = 1,\n page_size: int = 50\n) -> PageResult[Order]:\n orders, total = await db.get_orders_page(page, page_size)\n return PageResult.create(\n items=orders,\n page=page,\n page_size=page_size,\n total_items=total\n )\n```\n\nSee the [Pagination Guide](https://featureform.github.io/enrichmcp/pagination) for more examples.\n\n### \ud83d\udd10 Context & Authentication\n\nPass auth, database connections, or any context:\n\n```python\nfrom pydantic import Field\nfrom enrichmcp import EnrichModel\n\nclass UserProfile(EnrichModel):\n \"\"\"User profile information.\"\"\"\n\n user_id: int = Field(description=\"User ID\")\n bio: str | None = Field(default=None, description=\"Short bio\")\n\n@app.retrieve\nasync def get_user_profile(user_id: int, context: EnrichContext) -> UserProfile:\n # Access context provided by MCP client\n auth_user = context.get(\"authenticated_user_id\")\n if auth_user != user_id:\n raise PermissionError(\"Can only access your own profile\")\n return await db.get_profile(user_id)\n```\n\n### \u26a1 Request Caching\n\nReduce API overhead by storing results in a per-request, per-user, or global cache:\n\n```python\n\n@app.retrieve\nasync def get_customer(cid: int, ctx: EnrichContext) -> Customer:\n async def fetch() -> Customer:\n return await db.get_customer(cid)\n\n return await ctx.cache.get_or_set(f\"customer:{cid}\", fetch)\n```\n\n### \ud83e\udded Parameter Hints\n\nProvide examples and metadata for tool parameters using `EnrichParameter`:\n\n```python\nfrom enrichmcp import EnrichParameter\n\n@app.retrieve\nasync def greet_user(name: str = EnrichParameter(description=\"user name\", examples=[\"bob\"])) -> str:\n return f\"Hello {name}\"\n```\n\nTool descriptions will include the parameter type, description, and examples.\n\n### \ud83c\udf10 HTTP & SSE Support\n\nServe your API over standard output (default), SSE, or HTTP:\n\n```python\napp.run() # stdio default\napp.run(transport=\"streamable-http\")\n```\n\n## Why EnrichMCP?\n\nEnrichMCP adds three critical layers on top of MCP:\n\n1. **Semantic Layer** - AI agents understand what your data means, not just its structure\n2. **Data Layer** - Type-safe models with validation and relationships\n3. **Control Layer** - Authentication, pagination, and business logic\n\nThe result: AI agents can work with your data as naturally as a developer using an ORM.\n\n## Examples\n\nCheck out the [examples directory](examples/README.md):\n\n- [hello_world](examples/hello_world) - The smallest possible EnrichMCP app\n- [hello_world_http](examples/hello_world_http) - HTTP example using streamable HTTP\n- [shop_api](examples/shop_api) - In-memory shop API with pagination and filters\n- [shop_api_sqlite](examples/shop_api_sqlite) - SQLite-backed version\n- [shop_api_gateway](examples/shop_api_gateway) - EnrichMCP as a gateway in front of FastAPI\n- [sqlalchemy_shop](examples/sqlalchemy_shop) - Auto-generated API from SQLAlchemy models\n- [mutable_crud](examples/mutable_crud) - Demonstrates mutable fields and CRUD decorators\n- [caching](examples/caching) - Demonstrates ContextCache usage\n- [basic_memory](examples/basic_memory) - Simple note-taking API using FileMemoryStore\n- [openai_chat_agent](examples/openai_chat_agent) - Interactive chat client for MCP examples\n\n## Documentation\n\n- \ud83d\udcd6 [Full Documentation](https://featureform.github.io/enrichmcp)\n- \ud83d\ude80 [Getting Started Guide](https://featureform.github.io/enrichmcp/getting-started)\n- \ud83d\udd27 [API Reference](https://featureform.github.io/enrichmcp/api)\n\n## Contributing\n\nWe welcome contributions! See [CONTRIBUTING.md](CONTRIBUTING.md) for details.\n\n## Development Setup\n\nThe repository requires **Python 3.11** or newer. The Makefile includes\ncommands to create a virtual environment and run the tests:\n\n```bash\nmake setup # create .venv and install dependencies\nsource .venv/bin/activate\nmake test # run the test suite\n```\n\nThis installs all development extras and pre-commit hooks so commands like\n`make lint` or `make docs` work right away.\n\n## License\n\nApache 2.0 - See [LICENSE](LICENSE)\n\n---\n\nBuilt by [Featureform](https://featureform.com) \u2022 [MCP Protocol](https://modelcontextprotocol.io)\n",
"bugtrack_url": null,
"license": null,
"summary": "Transform Your Data Model into an MCP API",
"version": "0.4.6",
"project_urls": {
"Changelog": "https://github.com/featureform/enrichmcp/releases",
"Discussions": "https://github.com/featureform/enrichmcp/discussions",
"Documentation": "https://featureform.com/enrichmcp",
"Homepage": "https://github.com/featureform/enrichmcp",
"Issues": "https://github.com/featureform/enrichmcp/issues",
"Repository": "https://github.com/featureform/enrichmcp"
},
"split_keywords": [
"mcp",
" ai",
" agents",
" llm",
" model-context-protocol",
" pydantic",
" api",
" data-model",
" orm",
" agentic",
" enrichment"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "b7c8646a6b6889e19d920f31814aa32a18112ff976441f20fe1627db3ff1d5ed",
"md5": "2bb392b9f463d61e3fd38a60060ccf21",
"sha256": "fbaf7fd9de3a6bbd0f3f95568b436fb8842315c311a68340c060de749cb71536"
},
"downloads": -1,
"filename": "enrichmcp-0.4.6-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2bb392b9f463d61e3fd38a60060ccf21",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 34276,
"upload_time": "2025-07-10T10:20:40",
"upload_time_iso_8601": "2025-07-10T10:20:40.555817Z",
"url": "https://files.pythonhosted.org/packages/b7/c8/646a6b6889e19d920f31814aa32a18112ff976441f20fe1627db3ff1d5ed/enrichmcp-0.4.6-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "ee1f681ed55003f8ce2064931ff05d6c5c2e85dcda6d4dda3dcb34d92c14a9b8",
"md5": "aebb27aa397a7fe2b4c27ee60f03a2ac",
"sha256": "df0baf371141b0f8e7f6f8179396787c7d487f5c119ae8b070bd2034f8e4303d"
},
"downloads": -1,
"filename": "enrichmcp-0.4.6.tar.gz",
"has_sig": false,
"md5_digest": "aebb27aa397a7fe2b4c27ee60f03a2ac",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 118172,
"upload_time": "2025-07-10T10:20:41",
"upload_time_iso_8601": "2025-07-10T10:20:41.941442Z",
"url": "https://files.pythonhosted.org/packages/ee/1f/681ed55003f8ce2064931ff05d6c5c2e85dcda6d4dda3dcb34d92c14a9b8/enrichmcp-0.4.6.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-10 10:20:41",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "featureform",
"github_project": "enrichmcp",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"requirements": [
{
"name": "pydantic",
"specs": [
[
">=",
"2.0.0"
],
[
"<",
"3.0.0"
]
]
},
{
"name": "mcp-python",
"specs": [
[
">=",
"0.1.0"
]
]
},
{
"name": "typing-extensions",
"specs": [
[
">=",
"4.5.0"
],
[
"<",
"5.0.0"
]
]
}
],
"lcname": "enrichmcp"
}
JSON
