EnrichMCP：为AI代理提供数据模型访问框架

在当今数字化的时代，人工智能（AI）技术的发展日新月异。AI代理在各个领域的应用越来越广泛，如何让AI代理更好地理解和处理数据成为了一个关键问题。EnrichMCP作为一个Python框架，为解决这个问题提供了有效的方案。下面，我们就来详细了解一下EnrichMCP。

一、EnrichMCP概述

1.1 什么是EnrichMCP？

简单来说，EnrichMCP就像是SQLAlchemy对于AI代理的存在。它是一个基于模型上下文协议（MCP）构建的Python框架，其主要作用是帮助AI代理理解和处理你的数据。通过为数据模型添加一个语义层，EnrichMCP可以将你的数据模型转化为类型化、可发现的工具，就像是一个专门为AI设计的对象关系映射（ORM）工具。

1.2 EnrichMCP的主要目标

EnrichMCP有几个明确的目标：

生成类型化工具：从你的数据模型中自动生成类型化的工具，让AI代理可以更准确地与数据交互。
管理实体关系：处理实体之间的关系，例如用户与订单、订单与产品之间的关系，使AI代理能够自然地在这些关系之间进行导航。
提供模式发现：为AI代理提供数据结构的发现功能，让它们能够了解整个数据模型的结构。
输入输出验证：使用Pydantic模型对所有输入和输出进行验证，确保数据的有效性和一致性。
支持任意后端：可以与任何后端数据源一起工作，包括数据库、API或自定义逻辑。

二、EnrichMCP项目结构

2.1 项目布局

EnrichMCP项目有一个清晰的布局，各个部分分工明确，方便开发者进行开发和维护。以下是项目的主要目录结构：

/ (repo root)
├── README.md          – 项目介绍和快速入门指南
├── Makefile           – 包含常用的开发命令
├── pyproject.toml     – 项目元数据和工具配置
├── docs/              – 用户文档（使用MkDocs生成的网站）
├── examples/          – 可运行的示例代码
├── src/enrichmcp/     – 库的实现代码
└── tests/             – 单元测试代码

2.2 重要文件及其作用

2.2.1 pyproject.toml

这个文件定义了项目的元数据，包括所需的Python版本、依赖项、可选的开发工具以及工具配置，如Ruff、Pyright和代码覆盖率设置等。

2.2.2 Makefile

它包含了一些常用的开发任务，例如设置开发环境、代码检查、运行测试、生成文档等。例如，运行make setup命令可以创建一个虚拟环境并安装项目所需的依赖项。

2.2.3 docs/

这个目录包含了用户文档，使用Markdown格式编写。文档中详细描述了EnrichMCP的核心概念、使用示例、分页策略以及SQLAlchemy集成等内容，并且可以通过MkDocs将这些文档转换为一个网站进行浏览。

2.2.4 src/enrichmcp/

这是EnrichMCP框架的核心实现代码所在的目录。下面是一些主要的模块及其功能：

app.py：定义了EnrichMCP类，是框架的主要应用类。在初始化时，它会存储元数据并创建一个FastMCP实例。同时，它还提供了一些重要的功能，如数据模型探索资源、实体注册装饰器、模型描述生成方法等。
entity.py：定义了EnrichModel类，它是一个基于Pydantic的BaseModel，并添加了一些额外的辅助功能。例如，它可以检测关系字段并在序列化输出时将其移除，以避免递归问题；还可以返回一个Markdown格式的模型描述。
relationship.py：实现了Relationship描述符，用于声明实体之间的关系。它支持解析器的注册，并会自动创建具有描述性名称和文档字符串的MCP资源。同时，它还会验证解析器的返回类型是否与注释的关系类型匹配。
pagination.py：定义了分页结果类型，如PageResult用于基于页码的分页，CursorResult用于基于游标（cursor）的分页。此外，还提供了一些辅助参数类，用于在资源和解析器中实现分页功能。
context.py：是对FastMCP上下文对象的一个薄包装。
lifespan.py：提供了一个辅助函数，用于组合异步生命周期函数。
sqlalchemy/：包含了与SQLAlchemy集成的工具，如将SQLAlchemy模型自动转换为EnrichMCP实体，并注册默认资源和关系解析器。

2.2.5 examples/

这个目录包含了一些可运行的示例代码，展示了EnrichMCP的不同使用场景，如简单的Hello World API、基于内存和SQLite的商店API、OpenAI聊天代理以及SQLAlchemy集成等。

2.2.6 tests/

这里包含了单元测试代码，用于验证EnrichMCP应用的行为，包括实体注册、资源装饰、分页工具和模型描述生成等功能。

三、EnrichMCP的核心功能

3.1 EnrichMCP应用

3.1.1 初始化和元数据存储

src/enrichmcp/app.py中定义的EnrichMCP类在初始化时会存储元数据，并创建一个FastMCP实例。这个实例将作为整个应用的核心，负责处理各种请求和管理资源。

3.1.2 数据模型探索资源

EnrichMCP提供了一个内置的资源explore_data_model，通过调用这个资源，AI代理可以获得所有已注册实体、关系和使用提示的全面描述。这使得AI代理能够快速了解整个数据模型的结构和使用方法。

3.1.3 实体注册

EnrichMCP使用entity装饰器来注册实体。在注册过程中，它会验证模型是否有描述，并且所有字段（除了关系字段）是否包含Field(..., description="...")。注册的关系会被附加到模型类上，以便后续解析。

3.1.4 模型描述生成

describe_model方法可以生成每个实体、其字段和关系的文档描述。这对于AI代理和开发者来说都非常有用，他们可以通过这些文档了解数据模型的详细信息。

3.1.5 资源注册

resource装饰器用于将函数包装为MCP资源。在注册时，它要求函数有描述，并将其注册到FastMCP中。这样，AI代理就可以通过调用这些资源来获取数据或执行操作。

3.1.6 运行前验证

在运行服务器之前，run()方法会验证所有关系是否至少有一个解析器。如果没有，则会抛出ValueError异常，以确保数据模型的完整性和可用性。

3.2 实体和关系

3.2.1 实体（EnrichModel）

src/enrichmcp/entity.py中定义的EnrichModel是一个基于Pydantic的BaseModel，它添加了一些额外的功能。例如，relationship_fields()方法可以检测关系字段，并在序列化输出时将其移除，以避免递归问题。describe()方法可以返回一个Markdown格式的描述，包括字段和关系的信息。

3.2.2 关系（Relationship）

src/enrichmcp/relationship.py中实现的Relationship描述符用于声明实体之间的关系。它支持解析器的注册，通过@Entity.field.resolver装饰器可以为关系字段注册解析器。解析器会自动创建具有描述性名称和文档字符串的MCP资源，并且会验证解析器的返回类型是否与注释的关系类型匹配。

3.3 分页工具

src/enrichmcp/pagination.py定义了一些分页结果类型和辅助参数类，用于处理大数据集的分页问题。

PageResult：用于基于页码的分页。它包含了计算has_previous和total_pages的方法，方便开发者实现分页逻辑。
CursorResult：用于基于游标（cursor）的分页。它可以检测是否有下一页，并提供下一页的参数。
PaginationParams和CursorParams：提供了一些辅助参数，用于在资源和解析器中实现分页功能。

3.4 SQLAlchemy集成

EnrichMCP提供了与SQLAlchemy的集成功能，方便开发者将现有的SQLAlchemy模型转换为EnrichMCP实体。sqlalchemy包中包含了一些辅助工具，如EnrichSQLAlchemyMixin、include_sqlalchemy_models和sqlalchemy_lifespan，用于数据库的设置和模型的转换。

四、EnrichMCP的安装和使用

4.1 安装

你可以使用pip来安装EnrichMCP。如果你只需要基本功能，可以使用以下命令：

pip install enrichmcp

如果你需要使用SQLAlchemy集成功能，可以使用以下命令：

pip install enrichmcp[sqlalchemy]

4.2 使用示例

4.2.1 已有SQLAlchemy模型

如果你已经有了SQLAlchemy模型，你可以通过以下步骤将它们转换为AI可导航的API：

from enrichmcp import EnrichMCP
from enrichmcp.sqlalchemy import include_sqlalchemy_models, sqlalchemy_lifespan, EnrichSQLAlchemyMixin
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")

# 添加mixin到你的声明式基类
class Base(DeclarativeBase, EnrichSQLAlchemyMixin):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)
    status: Mapped[str] = mapped_column(default="active")
    orders: Mapped[list["Order"]] = relationship(back_populates="user")

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    total: Mapped[float] = mapped_column()
    user: Mapped[User] = relationship(back_populates="orders")

# 创建MCP应用
app = EnrichMCP(
    "E-commerce Data",
    lifespan=sqlalchemy_lifespan(Base, engine, cleanup_db_file=True),
)
include_sqlalchemy_models(app, Base)

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

在这个示例中，我们首先创建了一个异步的SQLAlchemy引擎，然后定义了User和Order两个模型，并添加了EnrichSQLAlchemyMixin。接着，我们创建了一个EnrichMCP应用，并使用include_sqlalchemy_models方法将SQLAlchemy模型集成到应用中。最后，我们运行应用。

4.2.2 已有REST API

如果你有现有的REST API，你可以使用EnrichMCP来为它们添加语义理解。以下是一个示例：

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

app = EnrichMCP("API Gateway")

@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: str = Field(description="Subscription tier: free, pro, enterprise")

    # 定义可导航的关系
    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: str = Field(description="Order status: pending, shipped, delivered")

    customer: Customer = Relationship(description="Customer who placed this order")

# 定义如何获取数据
@app.resource
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())

# 定义关系解析器
@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()]

app.run()

在这个示例中，我们首先创建了一个EnrichMCP应用，然后定义了Customer和Order两个实体，并定义了它们之间的关系。接着，我们定义了一个资源get_customer用于从CRM API中获取客户信息，以及一个关系解析器get_customer_orders用于获取客户的订单信息。最后，我们运行应用。

4.2.3 完全自定义控制

如果你想完全自定义数据层，可以使用以下示例：

from enrichmcp import EnrichMCP, EnrichModel, Relationship, EnrichContext
from datetime import datetime
from decimal import Decimal

app = EnrichMCP("Analytics Platform")

@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")

    # 计算字段
    lifetime_value: Decimal = Field(description="Total revenue from user")
    churn_risk: float = Field(description="ML-predicted churn probability 0-1")

    # 关系
    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.resource
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]

# 异步计算字段解析器
@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))

# 机器学习驱动的字段
@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()

在这个示例中，我们创建了一个EnrichMCP应用，并定义了User和Segment两个实体，以及它们之间的关系。我们还定义了一个复杂的资源find_high_value_at_risk_users，用于查找可能流失的高价值客户。此外，我们还为User实体的计算字段lifetime_value和churn_risk定义了解析器。最后，我们运行应用。

五、EnrichMCP的关键特性

5.1 自动模式发现

AI代理可以通过一次调用explore_data_model()方法来探索整个数据模型。这个方法会返回一个包含所有实体、字段、类型和关系的完整模式，让AI代理能够快速了解数据的结构。

5.2 关系导航

EnrichMCP允许你定义实体之间的关系，AI代理可以自然地在这些关系之间进行导航。例如，AI可以通过user → orders → products → categories这样的路径来访问相关数据。

5.3 类型安全和验证

EnrichMCP使用Pydantic进行全面的输入输出验证，确保数据的有效性和一致性。例如，你可以在定义实体字段时指定数据类型和验证规则，如total: float = Field(ge=0, description="Must be positive")，这样可以确保输入的total值是一个非负的浮点数。

5.4 可变性和CRUD操作

实体字段默认是不可变的，但你可以将它们标记为可变，并使用自动生成的补丁模型进行更新。同时，EnrichMCP还支持创建、更新和删除操作，你可以通过定义相应的资源函数来实现这些功能。

5.5 内置分页功能

EnrichMCP提供了内置的分页功能，用于处理大数据集。你可以使用PageResult和CursorResult等类型来实现基于页码或游标（cursor）的分页。例如：

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
    )

更多分页示例可以参考Pagination Guide。

5.6 上下文和认证

EnrichMCP允许你传递认证信息、数据库连接或任何上下文。例如，在资源函数中，你可以通过context参数访问上下文信息，实现认证和权限控制。

@app.resource
async def get_user_profile(user_id: int, context: EnrichContext) -> UserProfile:
    # 访问MCP客户端提供的上下文
    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)

六、EnrichMCP的优势

6.1 语义层

EnrichMCP为数据模型添加了一个语义层，使得AI代理不仅能够了解数据的结构，还能够理解数据的含义。这有助于AI代理更准确地处理和分析数据。

6.2 数据层

通过使用类型安全的模型和验证机制，EnrichMCP确保了数据的一致性和有效性。同时，它还支持实体之间的关系管理，使得数据的组织和访问更加自然和方便。

6.3 控制层

EnrichMCP提供了认证、分页和业务逻辑等控制功能，让开发者可以更好地管理和控制AI代理与数据的交互。这使得AI代理能够像开发者使用ORM一样自然地与数据进行交互。

七、EnrichMCP的示例和文档

7.1 示例

EnrichMCP的examples目录中包含了一些可运行的示例，展示了不同的使用场景：

hello_world：最小的EnrichMCP应用示例。
shop_api：基于内存的商店API示例，包含分页和过滤功能。
shop_api_sqlite：基于SQLite的商店API示例。
shop_api_gateway：将EnrichMCP作为FastAPI前端网关的示例。
sqlalchemy_shop：从SQLAlchemy模型自动生成API的示例。

7.2 文档

EnrichMCP提供了详细的文档，帮助开发者快速上手和使用：

Full Documentation：完整的文档。
Getting Started Guide：入门指南。
API Reference：API参考文档。

八、常见问题解答（FAQ）

8.1 EnrichMCP可以与哪些后端数据源一起使用？

EnrichMCP可以与任何后端数据源一起使用，包括数据库、API或自定义逻辑。它提供了与SQLAlchemy的集成功能，方便你将现有的SQLAlchemy模型转换为EnrichMCP实体。

8.2 如何安装EnrichMCP？

你可以使用pip来安装EnrichMCP。如果你只需要基本功能，可以使用pip install enrichmcp；如果你需要使用SQLAlchemy集成功能，可以使用pip install enrichmcp[sqlalchemy]。

8.3 如何定义实体和关系？

你可以使用@app.entity装饰器来定义实体，使用Relationship描述符来定义实体之间的关系。例如：

@app.entity
class Customer(EnrichModel):
    id: int = Field(description="Unique customer ID")
    orders: list["Order"] = Relationship(description="Customer's purchase history")

@app.entity
class Order(EnrichModel):
    id: int = Field(description="Order ID")
    customer: Customer = Relationship(description="Customer who placed this order")

8.4 如何实现分页功能？

EnrichMCP提供了PageResult和CursorResult等类型，用于实现基于页码或游标（cursor）的分页。你可以在资源函数中使用这些类型来返回分页结果。例如：

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
    )

8.5 如何进行输入输出验证？

EnrichMCP使用Pydantic进行输入输出验证。你可以在定义实体字段时指定数据类型和验证规则，如total: float = Field(ge=0, description="Must be positive")，这样可以确保输入的total值是一个非负的浮点数。

九、总结

EnrichMCP是一个强大的Python框架，它为AI代理提供了一种结构化的方式来访问和处理数据。通过添加语义层、数据层和控制层，EnrichMCP使得AI代理能够像开发者使用ORM一样自然地与数据进行交互。它支持多种后端数据源，提供了丰富的功能，如自动模式发现、关系导航、类型安全和验证、可变性和CRUD操作、内置分页功能以及上下文和认证等。同时，EnrichMCP还提供了详细的文档和丰富的示例，方便开发者快速上手和使用。如果你正在寻找一种方法来让AI代理更好地理解和处理你的数据，那么EnrichMCP绝对是一个值得考虑的选择。

EnrichMCP：AI代理数据模型访问的革命性Python框架