EnrichMCP
面向AI代理的ORM-将您的数据模型转换为语义MCP层
EnrichMCP是一个Python框架，可帮助AI代理理解和导航您的数据。它基于MCP（模型上下文协议）构建，添加了一个语义层，将您的数据模型转换为类型化的、可发现的工具，如人工智能的ORM。
什么是EnrichMCP？
将其视为AI代理的SQLAlchemy。EnrichMCP自动：
- 生成类型化工具 从您的数据模型
- 处理关系 实体（用户）之间→ 订单→ 产品）
- 提供架构发现 因此，AI代理能够理解您的数据结构
- 验证所有输入/输出 使用Pydantic模型
- 适用于任何后端 -数据库、API或自定义逻辑
安装
pip install enrichmcp
# With SQLAlchemy support
pip install enrichmcp[sqlalchemy]显示代码
选项1：我有SQLAlchemy模型（30秒）
将您现有的SQLAlchemy模型转换为AI可导航的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代理现在可以：
explore_data_model()-理解你的整个模式list_users(status='active')-使用筛选器进行查询get_user(id=123)-获取特定记录- 浏览关系：
user.orders→order.user
选项2：我有REST API（2分钟）
用语义理解来包装现有的API：
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()选项3：我想要完全控制（5分钟）
使用自定义逻辑构建完整的数据层：
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()主要特点
🔍 自动架构发现
AI代理只需一次调用即可探索您的整个数据模型：
schema = await explore_data_model()
# Returns complete schema with entities, fields, types, and relationships🔗 关系导航
一旦定义了关系，AI代理就会自然地遍历：
# AI can navigate: user → orders → products → categories
user = await get_user(123)
orders = await user.orders() # Automatic resolver
products = await orders[0].products()🛡️ 类型安全与验证
对每次交互进行完整的Pydantic验证：
@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() 将列出这些允许的值，以便代理知道有效的选项。
✏️ 可变性和CRUD
默认情况下，字段是不可变的。将它们标记为可变并使用 用于更新的自动生成补丁模型：
@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: ...📄 内置分页功能
优雅地处理大型数据集：
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)请参阅 分页指南 更多示例。
🔐 上下文和身份验证
传递身份验证、数据库连接或任何上下文：
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)⚡ 请求缓存
通过将结果存储在per-request、per-user或全局缓存中，减少API开销：
@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)🧭 参数提示
使用提供工具参数的示例和元数据 EnrichParameter:
from enrichmcp import EnrichParameter
@app.retrieve()
async def greet_user(name: str = EnrichParameter(description="user name", examples=["bob"])) -> str:
return f"Hello {name}"工具说明将包括参数类型、说明和示例。
🌐 HTTP和SSE支持
通过标准输出（默认）、SSE或HTTP为API提供服务：
app.run() # stdio default
app.run(transport="streamable-http")为什么选择EnrichMCP？
EnrichMCP在MCP之上增加了三个关键层：
- 语义层 -AI代理了解你的数据意味着什么，而不仅仅是它的结构
- 数据层 -具有验证和关系的类型安全模型
- 控制层 -身份验证、分页和业务逻辑
结果：AI代理可以像使用ORM的开发人员一样自然地处理您的数据。
服务器端LLM采样
EnrichMCP可以通过MCP请求语言模型完成 采样 功能。呼叫 ctx.ask_llm() 或 ctx.sampling() 来自任何资源的别名 并且连接的客户端将选择LLM并支付使用费。你可以调音 使用以下选项的行为 model_preferences, allow_tools，以及 max_tokens。参见 docs/server_side_llm.md 更多 细节。
示例
看看 示例目录:
- 你好_世界 -最小的EnrichMCP应用程序
- hello_world_http -使用流式HTTP的HTTP示例
- shop_api -带分页和过滤器的API内存车间
- shop_api_sqlite -SQLite支持版本
- shop_api_gateway -EnrichMCP作为FastAPI前的网关
- sqlalchemy_shop -从SQLAlchemy模型自动生成API
- mutable_crud -演示可变字段和CRUD装饰器
- 缓存 -演示ContextCache的使用方法
- 基本记忆 -使用FileMemoryStore进行简单的注释API
- openai_chat_agent -MCP示例的交互式聊天客户端
文档
贡献
我们欢迎捐款！看 贡献.md 了解详情。
开发设置
存储库需要 Python 3.11 或更新。Makefile包括 创建虚拟环境并运行测试的命令：
make setup # create .venv and install dependencies
source .venv/bin/activate
make test # run the test suite这将安装所有开发附加功能和预提交挂钩，因此命令如下 make lint 或 make docs 马上工作。
许可证
Apache 2.0-请参阅 许可证
______________________________________________________________________