What is Mcp Engine

MCPEngine is a production-grade implementation of the Model Context Protocol (MCP), designed for enterprise and real-world remote applications. It provides a secure and scalable framework for exposing data and tools to Large Language Models (LLMs) via MCP.

Use cases

Use cases for MCPEngine include integrating LLMs with applications like Slack, Gmail, and GitHub, enabling standardized endpoints for LLM access, and facilitating secure data exchanges in enterprise systems.

How to use

To use MCPEngine, integrate it with your applications by setting up a local proxy to connect with LLM hosts. Utilize the built-in OAuth for secure authentication and configure scope-based authorization for tools and resources.

Key features

Key features of MCPEngine include built-in OAuth support with various identity providers, an HTTP-first design utilizing Server-Sent Events (SSE), scope-based authorization, seamless bridging for LLM hosts, and full backwards compatibility with FastMCP and the official MCP SDK.

Where to use

MCPEngine can be used in various fields such as enterprise applications, cloud-native environments, and any scenario requiring integration with Large Language Models for enhanced data processing and interaction.

EnrichMCP

The ORM for AI Agents - Turn your data model into a semantic MCP layer

EnrichMCP is a Python framework that helps AI agents understand and navigate your data. Built on MCP (Model Context Protocol), it adds a semantic layer that turns your data model into typed, discoverable tools - like an ORM for AI.

What is EnrichMCP?

Think of it as SQLAlchemy for AI agents. EnrichMCP automatically:

• Generates typed tools from your data models
• Handles relationships between entities (users → orders → products)
• Provides schema discovery so AI agents understand your data structure
• Validates all inputs/outputs with Pydantic models
• Works with any backend - databases, APIs, or custom logic

Installation

pip install enrichmcp

# With SQLAlchemy support
pip install enrichmcp[sqlalchemy]

Show Me Code

Option 1: I Have SQLAlchemy Models (30 seconds)

Transform your existing SQLAlchemy models into an AI-navigable 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")

# Add the mixin to your declarative base
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")

# That's it! Create your MCP app
app = EnrichMCP(
    "E-commerce Data",
    lifespan=sqlalchemy_lifespan(Base, engine, cleanup_db_file=True),
)
include_sqlalchemy_models(app, Base)

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

AI agents can now:

• explore_data_model() - understand your entire schema
• list_users(status='active') - query with filters
• get_user(id=123) - fetch specific records
• Navigate relationships: user.orders → order.user

Option 2: I Have REST APIs (2 minutes)

Wrap your existing APIs with semantic understanding:

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

    # 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: str = Field(description="Order status: pending, shipped, delivered")

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

app.run()

Option 3: I Want Full Control (5 minutes)

Build a complete data layer with custom logic:

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

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

# 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, 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()

Key Features

🔍 Automatic Schema Discovery

AI agents explore your entire data model with one call:

schema = await explore_data_model()
# Returns complete schema with entities, fields, types, and relationships

🔗 Relationship Navigation

Define relationships once, AI agents traverse naturally:

# AI can navigate: user → orders → products → categories
user = await get_user(123)
orders = await user.orders()  # Automatic resolver
products = await orders[0].products()

🛡️ Type Safety & Validation

Full Pydantic validation on every interaction:

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

✏️ Mutability & CRUD

Fields are immutable by default. Mark them as mutable and use
auto-generated patch models for updates:

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

📄 Pagination Built-in

Handle large datasets elegantly:

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
    )

See the Pagination Guide for more examples.

🔐 Context & Authentication

Pass auth, database connections, or any context:

@app.retrieve
async def get_user_profile(user_id: int, context: EnrichContext) -> UserProfile:
    # Access context provided by MCP client
    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)

Why EnrichMCP?

EnrichMCP adds three critical layers on top of MCP:

1. Semantic Layer - AI agents understand what your data means, not just its structure
2. Data Layer - Type-safe models with validation and relationships
3. Control Layer - Authentication, pagination, and business logic

The result: AI agents can work with your data as naturally as a developer using an ORM.

Examples

Check out the examples directory:

• hello_world - The smallest possible EnrichMCP app
• shop_api - In-memory shop API with pagination and filters
• shop_api_sqlite - SQLite-backed version
• shop_api_gateway - EnrichMCP as a gateway in front of FastAPI
• sqlalchemy_shop - Auto-generated API from SQLAlchemy models
• mutable_crud - Demonstrates mutable fields and CRUD decorators
• openai_chat_agent - Interactive chat client for MCP examples

Documentation

• 📖 Full Documentation
• 🚀 Getting Started Guide
• 🔧 API Reference

Contributing

We welcome contributions! See CONTRIBUTING.md for details.

License

Apache 2.0 - See LICENSE

Built by Featureform • MCP Protocol