项目介绍
EnrichMCP
适用于AI代理的ORM - 将您的数据模型转换为语义化的MCP层
EnrichMCP是一个Python框架，帮助AI代理理解和导航您的数据。基于MCP（模型上下文协议）构建，它添加了一个语义层，将您的数据模型转化为类型化且可发现的工具——就像AI的ORM一样。
EnrichMCP是什么？
可以将其视为AI代理的SQLAlchemy。EnrichMCP自动：
- 从您的数据模型生成类型化工具
- 处理实体之间的关系（用户→订单→产品）
- 提供模式发现，使AI代理理解您的数据结构
- 使用Pydantic模型验证所有输入/输出
- 与任何后端工作——数据库、API或自定义逻辑
安装
pip install enrichmcp
# 带有SQLAlchemy支持
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")
# 在声明性基类中添加混合类
class Base(DeclarativeBase, EnrichSQLAlchemyMixin):
pass
class User(Base):
"""用户账户。"""
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, info={"description": "唯一用户ID"})
email: Mapped[str] = mapped_column(unique=True, info={"description": "电子邮件地址"})
status: Mapped[str] = mapped_column(default="active", info={"description": "账户状态"})
orders: Mapped[list["Order"]] = relationship(back_populates="user", info={"description": "此用户的全部订单"})
class Order(Base):
"""客户订单。"""
__tablename__ = "orders"
id: Mapped[int] = mapped_column(primary_key=True, info={"description": "订单ID"})
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), info={"description": "拥有者用户ID"})
total: Mapped[float] = mapped_column(info={"description": "订单总额"})
user: Mapped[User] = relationship(back_populates="orders", info={"description": "下单的用户"})
# 创建您的MCP应用
app = EnrichMCP(
"电子商务数据",
"由SQLAlchemy模型生成的API",
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网关", "围绕现有REST API的包装器")
http = httpx.AsyncClient(base_url="https://api.example.com")
@app.entity
class Customer(EnrichModel):
"""我们CRM系统中的客户。"""
id: int = Field(description="唯一客户ID")
email: str = Field(description="主要联系电子邮件")
tier: Literal["free", "pro", "enterprise"] = Field(
description="订阅级别"
)
# 定义可导航的关系
orders: list["Order"] = Relationship(description="客户的购买历史")
@app.entity
class Order(EnrichModel):
"""来自我们电子商务平台的客户订单。"""
id: int = Field(description="订单ID")
customer_id: int = Field(description="关联客户")
total: float = Field(description="订单总额（美元）")
status: Literal["pending", "shipped", "delivered"] = Field(
description="订单状态"
)
customer: Customer = Relationship(description="下单的客户")
# 定义如何获取数据
@app.retrieve
async def get_customer(customer_id: int) -> Customer:
"""从CRM API获取客户。"""
response = await http.get(f"/api/customers/{customer_id}")
return Customer(**response.json())
# 定义关系解析器
@Customer.orders.resolver
async def get_customer_orders(customer_id: int) -> list[Order]:
"""获取客户的订单。"""
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:
"""获取订单的客户。"""
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("分析平台", "自定义分析API")
db = ... # 您的数据库连接
@app.entity
class User(EnrichModel):
"""具有计算分析字段的用户。"""
id: int = Field(description="用户ID")
email: str = Field(description="联系方式电子邮件")
created_at: datetime = Field(description="注册日期")
# 计算字段
lifetime_value: Decimal = Field(description="来自用户的总收入")
churn_risk: float = Field(description="ML预测的流失概率0-1")
# 关系
orders: list["Order"] = Relationship(description="购买历史")
segments: list["Segment"] = Relationship(description="营销细分市场")
@app.entity
class Segment(EnrichModel):
"""用于营销的动态用户细分市场。"""
name: str = Field(description="细分市场名称")
criteria: dict = Field(description="细分市场标准")
users: list[User] = Relationship(description="属于这个细分市场的用户")
@app.entity
class Order(EnrichModel):
"""简化订单记录。"""
id: int = Field(description="订单ID")
user_id: int = Field(description="拥有者用户ID")
total: Decimal = Field(description="订单总额")
@User.orders.resolver
async def list_user_orders(user_id: int) -> list[Order]:
"""获取用户的订单。"""
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]:
"""获取包括用户的细分市场。"""
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]:
"""列出细分市场中的用户。"""
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]
# 具有业务逻辑的复杂资源
@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]:
"""找到可能流失的有价值客户。"""
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]
# 异步计算字段解析器
@User.lifetime_value.resolver
async def calculate_lifetime_value(user_id: int) -> Decimal:
"""计算用户订单的总收入。"""
total = await db.query_single(
"SELECT SUM(total) FROM orders WHERE user_id = ?",
user_id
)
return Decimal(str(total or 0))
# 基于机器学习的字段
@User.churn_risk.resolver
async def predict_churn_risk(user_id: int) -> float:
"""运行流失预测模型。"""
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()
# 返回包含实体、字段、类型和关系的完整模式
🔗 关系导航
定义一次关系，AI代理自然遍历：
# AI可以导航：用户 → 订单 → 产品 → 分类
user = await get_user(123)
orders = await user.orders() # 自动解析器
products = await orders[0].products()
🛡️ 类型安全及验证
每次交互都进行全Pydantic验证：
@app.entity
class Order(EnrichModel):
total: float = Field(ge=0, description="必须是正数")
email: EmailStr = Field(description="客户电子邮件")
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="电子邮件")
@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_id: int = Field(description="用户ID")
bio: str | None = Field(default=None, description="简短简介")
@app.retrieve
async def get_user_profile(user_id: int) -> UserProfile:
ctx = app.get_context()
# 访问由MCP客户端提供的上下文
auth_user = ctx.get("authenticated_user_id")
if auth_user != user_id:
raise PermissionError("只能访问自己的个人资料")
return await db.get_profile(user_id)
⚡ 请求缓存
通过在每个请求、每个用户或全局缓存中存储结果来减少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="用户名", examples=["bob"])) -> str:
return f"Hello {name}"
工具描述将包括参数类型、描述和示例。
🌐 HTTP及SSE支持
通过标准输出（默认）、SSE或HTTP提供您的API：
app.run() # 默认stdio
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。
示例
查看示例目录：
- hello_world - 最小的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装饰器
- caching - 展示ContextCache的使用
- basic_memory - 使用FileMemoryStore的简单笔记API
- openai_chat_agent - 用于MCP示例的交互式聊天客户端
文档
贡献
我们欢迎贡献！请参阅CONTRIBUTING.md了解详情。
开发环境设置
该仓库需要Python 3.11或更高版本。Makefile中包含了创建虚拟环境和运行测试的命令：
make setup # 创建.venv并安装依赖项
source .venv/bin/activate
make test # 运行测试套件
这将安装所有开发额外项和pre-commit钩子，因此命令如make lint或make docs可以立即运行。
许可证
Apache 2.0 - 查看LICENSE
由Featureform构建 • MCP协议