EnrichMCP – AI 에이전트를 위한 데이터 ORM Python 프레임워크

# 기본 설치
pip install enrichmcp

# SQLAlchemy 지원 포함
pip install enrichmcp[sqlalchemy]

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

# 선언적 베이스에 믹스인 추가
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()

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

app = EnrichMCP("API Gateway")

@app.entity
class Customer(EnrichModel):
    """CRM 시스템의 고객 정보."""

    id: int = Field(description="고유 고객 ID")
    email: str = Field(description="주 연락처 이메일")
    tier: str = Field(description="구독 등급: free, pro, enterprise")

    # 탐색 가능한 관계 정의
    orders: list["Order"] = Relationship(description="고객의 구매 이력")

@app.entity
class Order(EnrichModel):
    """전자상거래 플랫폼의 고객 주문."""

    id: int = Field(description="주문 ID")
    customer_id: int = Field(description="연결된 고객")
    total: float = Field(description="USD 단위 주문 총액")
    status: str = Field(description="주문 상태: pending, shipped, delivered")

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

app.run()

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

app = EnrichMCP("Analytics Platform")

@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.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, context: EnrichContext) -> float:
    """이탈 예측 모델을 실행합니다."""
    features = await gather_user_features(user_id)
    model = context.get("ml_models")["churn"]
    return float(model.predict_proba(features)[0][1])

app.run()

schema = await explore_data_model()
# 엔티티, 필드, 타입, 관계가 포함된 완전한 스키마 반환

# AI가 다음과 같이 탐색할 수 있습니다: user → orders → products → categories
user = await get_user(123)
orders = await user.orders()  # 자동 해결자
products = await orders[0].products()

@app.entity
class Order(EnrichModel):
    total: float = Field(ge=0, description="양수여야 함")
    email: EmailStr = Field(description="고객 이메일")
    status: Literal["pending", "shipped", "delivered"]

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

@app.retrieve
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("자신의 프로필만 접근할 수 있습니다")
    return await db.get_profile(user_id)