MCPEngine

EnrichMCP

AI エージェントのための ORM — データモデルをセマンティックな MCP レイヤーに変換

EnrichMCP は、AI エージェントがデータを理解し、データ内を探索できるようにする Python フレームワークです。MCP（Model Context Protocol）を基盤とし、データモデルを型付きで発見可能なツールに変換するセマンティックレイヤーを追加します。AI 向けの ORM のようなものです。

EnrichMCP とは？

AI エージェント向けの SQLAlchemy と考えてください。EnrichMCP は次の処理を自動で行います。

• データモデルから型付きツールを生成
• エンティティ間のリレーションシップを処理（users → orders → products）
• AI エージェントがデータ構造を理解できるよう、スキーマの探索機能を提供
• Pydantic モデルですべての入力／出力を検証
• データベース、API、カスタムロジックなど、あらゆるバックエンドで動作

インストール

pip install enrichmcp

# With SQLAlchemy support
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")


# Add the mixin to your declarative base
class Base(DeclarativeBase, EnrichSQLAlchemyMixin):
    pass


class User(Base):
    """User account."""

    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, info={"description": "Unique user ID"})
    email: Mapped[str] = mapped_column(unique=True, info={"description": "Email address"})
    status: Mapped[str] = mapped_column(default="active", info={"description": "Account status"})
    orders: Mapped[list["Order"]] = relationship(
        back_populates="user", info={"description": "All orders for this user"}
    )


class Order(Base):
    """Customer order."""

    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True, info={"description": "Order ID"})
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id"), info={"description": "Owner user ID"}
    )
    total: Mapped[float] = mapped_column(info={"description": "Order total"})
    user: Mapped[User] = relationship(
        back_populates="orders", info={"description": "User who placed the order"}
    )


# That's it! Create your MCP app
app = EnrichMCP(
    "E-commerce Data",
    "API generated from SQLAlchemy models",
    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 Gateway", "Wrapper around existing REST APIs")
http = httpx.AsyncClient(base_url="https://api.example.com")


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

    # Define navigable relationships
    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: Literal["pending", "shipped", "delivered"] = Field(description="Order status")

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


# Define how to fetch data
@app.retrieve()
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())


# Define relationship resolvers
@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()]


@Order.customer.resolver
async def get_order_customer(order_id: int) -> Customer:
    """Fetch the customer for an order."""
    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 = ...  # your database connection


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

    # Computed fields
    lifetime_value: Decimal = Field(description="Total revenue from user")
    churn_risk: float = Field(description="ML-predicted churn probability 0-1")

    # Relationships
    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.entity()
class Order(EnrichModel):
    """Simplified order record."""

    id: int = Field(description="Order ID")
    user_id: int = Field(description="Owner user ID")
    total: Decimal = Field(description="Order total")


@User.orders.resolver
async def list_user_orders(user_id: int) -> list[Order]:
    """Fetch orders for a user."""
    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]:
    """Fetch segments that include the user."""
    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]:
    """List users in a segment."""
    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]


# Complex resource with business logic
@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]:
    """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]


# Async computed field resolver
@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))


# ML-powered field
@User.churn_risk.resolver
async def predict_churn_risk(user_id: int) -> float:
    """Run churn prediction model."""
    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 エージェントは、1 回の呼び出しでデータモデル全体を探索できます。

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"]

describe_model() はこれらの許可された値を一覧表示し、エージェントが有効な選択肢を把握できるようにします。

✏️ 可変性と CRUD

フィールドはデフォルトで不変です。フィールドを可変として指定し、 自動生成されるパッチモデルを使って更新できます。

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

    user_id: int = Field(description="User ID")
    bio: str | None = Field(default=None, description="Short bio")


@app.retrieve()
async def get_user_profile(user_id: int) -> UserProfile:
    ctx = app.get_context()
    # Access context provided by MCP client
    auth_user = ctx.get("authenticated_user_id")
    if auth_user != user_id:
        raise PermissionError("Can only access your own profile")
    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="user name", examples=["bob"])) -> str:
    return f"Hello {name}"

ツールの説明には、パラメーターの型、説明、例が含まれます。

🌐 HTTP と SSE のサポート

標準出力（デフォルト）、SSE、または HTTP 経由で API を提供できます。

app.run()  # stdio default
app.run(transport="streamable-http")

EnrichMCP を選ぶ理由

EnrichMCP は MCP に次の 3 つの重要なレイヤーを追加します。

1. セマンティックレイヤー — AI エージェントがデータの構造だけでなく、その意味を理解します
2. データレイヤー — 検証とリレーションシップを備えた型安全なモデル
3. 制御レイヤー — 認証、ページネーション、ビジネスロジック

その結果、AI エージェントは ORM を使う開発者のように、自然にデータを扱えるようになります。

サーバーサイド LLM サンプリング

EnrichMCP は、MCP のサンプリング機能を通じて言語モデルの補完をリクエストできます。任意のリソースから 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 - FastAPI の前段にゲートウェイとして EnrichMCP を配置する例
• 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            # create .venv and install dependencies
source .venv/bin/activate
make test             # run the test suite

これにより、開発用の追加パッケージと pre-commit フックがすべてインストールされ、make lint や make docs などのコマンドをすぐに実行できます。

ライセンス

Apache 2.0 - 詳細は LICENSE をご覧ください。

---

Featureform • MCP Protocol によって開発