AI智能体像ORM程序员一样自然地保存数据


ORM for AI Agents -将您的数据模型转换为语义MCP层

EnrichMCP是一个Python框架,可以帮助AI智能体理解和导航您的数据。基于MCP(模型上下文协议)构建,它添加了一个语义层,将您的数据模型转换为类型化的可扩展工具-就像AI的ORM一样。


什么是EnrichMCP?
可以将其视为AI代理的SQLAlchemy。EnrichMCP自动:

  • 从数据模型生成类型化工具
  • 处理实体之间的关系(用户→订单→产品)
  • 提供模式发现,以便AI代理了解您的数据结构
  • 使用Pydantic模型验证所有输入/输出
  • 可与任何后端-数据库、API或自定义逻辑一起使用

为什么选择EnrichMCP?
EnrichMCP在MCP之上增加了三个关键层:
语义层-人工智能代理理解您的数据意味着什么,而不仅仅是其结构
数据层-具有验证和关系的类型安全模型
控制层-身份验证、分页和业务逻辑

结果:AI智能体可以像使用ORM的开发人员一样自然地处理您的数据。


关键特征
自动模式发现
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"]


可扩展性CRUD️
默认情况下,字段是不可变的。将它们标记为可变的,并使用自动生成的补丁模型进行更新:

@app.entity
class Customer(EnrichModel):
    id: int = Field(description="ID")
    email: str = Field(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.resource
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
    )

更多示例请参见分页指南。

身份验证上下文
传递身份验证、数据库连接或任何上下文:

@app.resource
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)