EnrichMCP

当我们在开发 AI 代理应用时，一个反复出现的痛点是如何让 AI 理解并使用我们的数据。传统做法是编写大量文档、手动定义工具函数、穷举可能的查询模式——这些工作枯燥且极易出错。EnrichMCP 彻底改变了这一范式：它不再要求你去教 AI 如何使用数据，而是让你的数据模型教会 AI 理解自己。只需定义一次带描述的 Pydantic 模型和关系，EnrichMCP 就能自动生成类型安全、关系感知的 MCP API，AI 代理从此能像天生就理解你的业务领域一样，自然地发现、导航和查询数据。

项目基本信息

一、项目介绍

EnrichMCP 是一个构建在 FastMCP 之上的高级数据访问层，由 Featureform 团队开发。它的核心哲学是 Agentic Enrichment：“不是教 AI 了解你的 API，而是让 API 教 AI 了解自己。”

核心概念：Agentic Enrichment

传统 MCP 工具需要你手动指定每个工具的名称、参数、返回值和文档。当数据模型复杂、实体关系众多时，这种手工方式难以扩展。EnrichMCP 将这个过程自动化了：

1. 你定义带丰富描述的 Pydantic 数据模型（实体、字段、关系）。
2. 你为关键关系编写简单的解析器函数（如何从一个实体获取关联的另一实体）。
3. EnrichMCP 自动生成所有 MCP 工具，并为 AI 提供一个统一的 explore_data_model() 入口，AI 通过这个入口就能自省整个数据模型。

核心能力

二、核心优势

从“手动教 AI”到“数据自描述”

这是 EnrichMCP 最根本的优势。传统方式下，每次新增一个数据实体，你都需要手动编写对应的 MCP 工具、参数说明、返回值文档。EnrichMCP 将这一切压缩为“定义 Pydantic 模型+编写解析器”，剩下的工作全自动完成。这不仅减少开发量，更避免了文档和实现不一致的隐患。

关系遍历的自动生成

实体之间往往存在关联：客户有订单、订单有商品、商品有分类。EnrichMCP 的 Relationship 让 AI 能自然地沿着这些关系导航。AI 查出客户后，可以自动“跟随” orders 关系获取其订单，再“跟随” items 关系获取订单详情——这一切无需 AI 提前知道关系存在，explore_data_model() 会告诉它路怎么走。

生产的就绪性

EnrichMCP 不是实验性玩具。它提供了错误处理（NotFoundError、ValidationError）、上下文管理（通过 EnrichContext 传递数据库连接和认证信息）、分页支持（基于页码和基于游标的分页），这些都是生产环境必备的能力。

Apache 2.0 开源协议

采用对商业集成友好的 Apache 2.0 协议，适合企业级使用。

三、适用场景

构建 AI 驱动的数据查询助手

用户：显示所有活跃客户及其最近 5 个订单。

AI 调用 explore_data_model() 了解 Customer 和 Order 的关系，然后调用 list_customers(status="active") 获取客户列表，再通过 Customer 的 orders 解析器获取关联订单。整个过程完全自动，无需 AI 提前知道有哪些字段和关系。

将已有数据库快速暴露给 AI 代理

如果你已经有一套定义好的 Pydantic/SQLAlchemy 模型，可以在几小时内将它们包装为 EnrichMCP API，让 AI 代理直接理解和查询数据。

多步骤、关系驱动的数据分析

当问题需要跨越多层关系时（例如：“找到过去一个月订购过特定商品的客户，列出他们的邮箱和订单总金额”），EnrichMCP 的关系导航能力能自然地支持这种复杂的图遍历。

快速原型验证与迭代

在数据模型还在频繁变化的早期阶段，EnrichMCP 的“定义即生成”模式能让 AI 代理始终同步最新的数据结构，无需反复修改工具定义。

四、安装教程

环境要求

安装步骤

一步安装

pip install enrichmcp

开发模式安装（如果有意贡献代码）

git clone https://github.com/featureform/enrichmcp.git
cd enrichmcp
make setup

运行测试

make test

五、使用示例

以下是一个完整示例，展示如何定义数据模型、注册关系和资源，然后运行 EnrichMCP 服务器。

步骤 1：定义数据模型

from enrichmcp import EnrichMCP, EnrichModel, Relationship
from pydantic import Field
from datetime import datetime

app = EnrichMCP(title="订单查询系统", description="AI 代理的客户与订单数据模型")

@app.entity
class Customer(EnrichModel):
    """客户实体，代表系统中的注册用户。

    包含核心信息并通过关系可导航至其订单。
    """
    id: int = Field(description="客户唯一标识符")
    name: str = Field(description="客户全名")
    email: str = Field(description="联系邮箱")
    status: str = Field(description="账户状态：活跃、暂停、流失")
    created_at: datetime = Field(description="注册时间")
    orders: list["Order"] = Relationship(description="该客户的订单列表")

@app.entity
class Order(EnrichModel):
    """订单实体，记录购买行为。

    可导航至下单客户和订单明细。
    """
    id: int = Field(description="订单唯一标识符")
    customer_id: int = Field(description="下单客户的 ID")
    total: float = Field(description="订单总金额")
    status: str = Field(description="状态：待处理、已发货、已交付")
    created_at: datetime = Field(description="下单时间")
    customer: Customer = Relationship(description="下此订单的客户")

步骤 2：实现关系解析器

@Customer.orders.resolver
async def get_customer_orders(customer_id: int) -> list[Order]:
    """从数据库获取某个客户的全部订单。"""
    return await db.fetch_orders_by_customer(customer_id)

@Order.customer.resolver
async def get_order_customer(order_id: int) -> Customer:
    """从数据库获取下此订单的客户信息。"""
    return await db.fetch_customer_by_order(order_id)

步骤 3：定义根资源入口

@app.resource
async def get_customer(customer_id: int) -> Customer:
    """通过 ID 查询客户。"""
    return await db.get_customer(customer_id)

@app.resource
async def list_customers(status: str | None = None) -> list[Customer]:
    """列出客户，可按状态筛选。"""
    return await db.list_customers(status=status)

步骤 4：启动服务器

if __name__ == "__main__":
    app.run()

AI 代理的实际交互流程

当 AI 连接到这个服务后：

1. 发现阶段：AI 自动调用 explore_data_model()，得知系统中有 Customer 和 Order 两个实体，它们之间存在双向关系。
2. 查询阶段：AI 调用 list_customers(status="活跃") 获取活跃客户列表。
3. 导航阶段：对于每个客户，AI 自动调用 Customer.orders 解析器，获取关联订单——不需要任何预先配置。

六、常见问题

问：EnrichMCP 和直接使用 FastMCP 有什么区别？

答：FastMCP 提供了 MCP 协议的底层实现，你需要手动定义每个工具。EnrichMCP 构建在 FastMCP 之上，提供了数据模型抽象层——你只需定义 Pydantic 模型和关系，工具自动生成。对于数据密集型应用，EnrichMCP 能减少大量样板代码。

问：如何处理数据库连接和认证？

答：通过 EnrichContext 传递上下文。你可以在上下文中包含数据库连接、认证令牌等，然后在解析器和资源函数中使用：

@app.resource
async def get_customer(customer_id: int, context: EnrichContext) -> Customer:
    return await context.db.get_customer(customer_id)

问：数据量大时如何避免返回海量数据？

答：EnrichMCP 内置了分页支持。你可以使用 PageResult（基于页码）或 CursorResult（基于游标）来处理大数据集，防止 AI 上下文窗口溢出。

问：如果 AI 请求不存在的实体会发生什么？

答：你可以使用内置的 NotFoundError 返回友好的 404 错误：

from enrichmcp.errors import NotFoundError
if not customer:
    raise NotFoundError(f"客户 {customer_id} 未找到")

问：支持哪些后端数据库？

答：EnrichMCP 本身不限制数据库类型。你需要在解析器和资源函数中自行实现数据库查询逻辑，可以连接任何 Python 支持的数据库（PostgreSQL、MySQL、MongoDB 等）。

七、总结

EnrichMCP 提出了一种新的 AI-数据交互范式：让数据模型自我描述，让 AI 自主导航。它用 Pydantic 的类型系统和装饰器，将数据层抽象提升到了一个新的高度——不再是手动编写工具，而是让工具从数据模型中自动“生长”出来。对于构建 AI 代理、数据查询助手或任何需要让 AI 理解结构化数据的应用，EnrichMCP 提供了一个务实、可扩展的解决方案。

花十分钟用 EnrichMCP 包装你的一个数据表，你会对“AI 理解数据”这件事产生全新的认知。