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]

示例代码

方法一：已有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

方法二：已有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("Analytics Platform", "Custom analytics 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))


# ML 기반 필드
@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])

주요 기능

🔍 자동 스키마 탐색

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()  # 标准输出默认
app.run(transport="streamable-http")

为什么选择 EnrichMCP？

EnrichMCP 在 MCP 的基础上增加了三层关键功能：

1. 语义层 - AI 代理不仅理解数据的结构，还能理解数据的实际含义
2. 数据层 - 类型安全的模型，具备验证和关系管理功能
3. 控制层 - 身份验证、分页和业务逻辑

结果是：AI 代理可以像开发者使用 ORM 一样自然地操作您的数据。

服务器端 LLM 采样

EnrichMCP 可以通过 MCP 的 sampling 功能请求语言模型完成任务。在任何资源中调用 ctx.ask_llm() 或其别名 ctx.sampling()，连接的客户端就会选择一个 LLM 并支付相应的费用。您可以通过 model_preferences、allow_tools 和 max_tokens 等选项来调整行为。更多详细信息请参阅 docs/server_side_llm.md。

示例

请查看 examples 目录：

• 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 示例的交互式聊天客户端

文档

• 📖 完整文档
• 🚀 入门指南
• 🔧 API 参考

贡献

我们欢迎贡献！详情请参阅 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 协议 构建

EnrichMCP 快速上手指南

EnrichMCP 是一个 Python 框架，旨在帮助 AI Agent 理解和导航数据。它基于 MCP（Model Context Protocol），将你的数据模型转换为具有语义层的、类型安全的可发现工具，类似于 AI 领域的 ORM。

环境准备

• Python 版本：需要 Python 3.11 或更高版本。
• 前置知识：了解 Pydantic 模型定义及基本的异步编程概念有助于更好地使用高级功能。
• 依赖建议：如果你计划连接关系型数据库，建议安装 SQLAlchemy 支持。

安装步骤

使用 pip 进行安装。根据你的需求选择以下命令之一：

基础安装：

pip install enrichmcp

包含 SQLAlchemy 支持（推荐用于数据库集成）：

pip install enrichmcp[sqlalchemy]

提示：如果下载速度较慢，可以使用国内镜像源加速，例如： pip install enrichmcp -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

EnrichMCP 提供了多种集成方式。以下是两种最常见场景的快速入门示例。

场景一：集成现有 SQLAlchemy 模型（最快上手）

如果你已经拥有 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")

# 将 Mixin 添加到声明基类中
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 Agent 现在可以执行的操作：

• explore_data_model()：理解整个数据结构。
• list_users(status='active')：带过滤条件的查询。
• get_user(id=123)：获取特定记录。
• 自动导航关系：如通过 user.orders 访问订单。

场景二：封装 REST API 或自定义逻辑

如果没有 ORM，或者需要封装现有的 HTTP API，可以使用 EnrichModel 手动定义实体和解析器。

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()]

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

核心特性说明：

• 类型安全：所有输入输出均通过 Pydantic 进行验证。
• 关系导航：定义一次关系，AI 即可自然遍历（如 customer.orders）。
• 模式发现：AI 可通过 explore_data_model 自动了解可用数据和字段含义。