关键词

AI代理数据接口语义化数据层数据模型工具LLM后端Python数据框架

项目简介

EnrichMCP是一个Python框架，旨在帮助AI代理更好地理解和导航您的数据。它基于Model Context Protocol (MCP) 构建，能够将您现有或自定义的数据模型转化为类型化、可发现的语义化工具，就像是AI的ORM（对象关系映射）。通过EnrichMCP，您可以轻松构建一个MCP服务器，向大型语言模型（LLM）客户端提供结构化的数据访问、功能调用和上下文信息。

主要功能点

  • 自动化工具生成: 从您的Python数据模型（如Pydantic模型或SQLAlchemy模型）自动生成可供AI代理调用的、带有类型定义的工具。
  • 关系管理: 自动处理实体之间（如用户到订单、订单到产品）的复杂关系，使AI能够自然地遍历数据。
  • Schema探索: 提供AI代理探索整个数据模型Schema的能力，帮助它们理解数据结构、字段类型和语义。
  • 类型安全与验证: 基于Pydantic模型对所有输入/输出进行严格的类型检查和数据验证。
  • 多后端支持: 能够与各种数据后端集成，包括数据库、REST API或自定义业务逻辑。
  • CRUD操作: 支持生成数据的创建（Create）、读取（Retrieve）、更新（Update）和删除（Delete）工具。
  • 内置分页: 为处理大量数据提供内置的分页（PageResult/CursorResult）机制。
  • 上下文与认证: 允许在请求上下文中传递认证信息、数据库连接或其他自定义数据。
  • 请求缓存: 提供多种粒度（请求、用户、全局）的缓存机制，减少数据源负载。
  • 服务器端LLM采样: 允许MCP服务器通过LLM客户端请求LLM完成（采样），实现更智能的业务逻辑。
  • 多种传输协议: 支持通过标准输出 (Stdio)、HTTP (Streamable HTTP) 等多种协议提供服务。

安装步骤

首先，请确保您的Python版本为3.11或更高。

  1. 基础安装: 
    pip install enrichmcp
  2. 带SQLAlchemy支持的安装 (如果您计划从SQLAlchemy模型自动生成API): 
    pip install enrichmcp[sqlalchemy]
  3. 开发环境设置 (如果您想为项目贡献或进行深度开发): 
    # 克隆仓库
git clone https://github.com/featureform/enrichmcp.git
cd enrichmcp
# 创建虚拟环境并安装依赖
make setup
source .venv/bin/activate

服务器配置 (MCP客户端使用)

MCP客户端需要配置服务器的启动命令和参数，以便连接到EnrichMCP服务器。以下是一个基本的JSON格式配置示例，您需要根据实际部署的EnrichMCP应用来调整 'command' 和 'args'：

{
  "mcpServers": {
    "your_server_name": {
      "command": "python",
      "args": [
        "/path/to/your/enrichmcp_app.py"
      ],
      "description": "这是您的EnrichMCP服务器的描述，例如：一个提供用户和订单数据的电商API。",
      "url": "可选：如果您的EnrichMCP应用运行在HTTP模式，可以提供URL，例如 'http://localhost:8000/mcp'"
    }
  }
}
  • 'your_server_name': 您为这个MCP服务器实例定义的名称，如 'E-commerce Data' 或 'Analytics Platform'。这个名称将在MCP客户端的配置中用作服务器的标识。
  • 'command': 启动Python解释器的命令，通常是 '"python"' 或您的虚拟环境中的Python路径。
  • 'args': 启动您的EnrichMCP应用程序的Python脚本的完整路径。例如，如果您有一个名为 'myapp.py' 的文件包含您的EnrichMCP应用定义，这里就填写其完整路径。
  • 'description': 对您的MCP服务器功能的简短说明，这将帮助LLM客户端理解其用途和可用能力。
  • 'url': 如果您的EnrichMCP应用程序通过HTTP（而不是默认的Stdio）运行，您可以在此处提供服务器的URL。例如，如果运行 'app.run(transport="streamable-http")'，则默认URL通常是 'http://localhost:8000/mcp'。

基本使用方法

以下是一个简单的EnrichMCP应用程序示例，展示了如何定义一个实体和一个检索资源，并通过 'app.run()' 启动服务器。

1. 创建您的EnrichMCP应用程序文件 (例如 'myapp.py'):

from enrichmcp import EnrichMCP, EnrichModel
from pydantic import Field
from datetime import datetime, UTC

# 创建EnrichMCP应用实例
app = EnrichMCP(
    title="My Simple API",
    instructions="提供用户数据的简单API",
)

# 定义一个实体（数据模型）
@app.entity(description="系统中的用户账户")
class User(EnrichModel):
    id: int = Field(description="用户唯一标识符")
    name: str = Field(description="用户全名")
    email: str = Field(description="用户邮箱地址")
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC), description="账户创建时间")

# 模拟数据存储
USERS_DB = {
    1: User(id=1, name="Alice Smith", email="[email protected]", created_at=datetime(2023, 1, 10, tzinfo=UTC)),
    2: User(id=2, name="Bob Johnson", email="[email protected]", created_at=datetime(2023, 3, 15, tzinfo=UTC)),
}

# 定义一个检索资源（工具）
@app.retrieve(description="根据ID获取单个用户")
async def get_user(user_id: int) -> User:
    """根据用户ID检索用户账户信息。"""
    user = USERS_DB.get(user_id)
    if user:
        return user
    # 返回一个表示未找到的User对象，或者抛出NotFoundError
    return User(id=0, name="Not Found", email="[email protected]", created_at=datetime(1970, 1, 1, tzinfo=UTC))

# 运行服务器
if __name__ == "__main__":
    print("启动My Simple API...")
    app.run()

2. 运行EnrichMCP服务器: 在终端中，导航到 'myapp.py' 所在的目录，然后执行：

python myapp.py

服务器将启动，并通过标准输出/输入（Stdio）等待MCP客户端连接。如果您希望通过HTTP运行，可以修改 'app.run()' 为 'app.run(transport="streamable-http")'。

3. 从MCP客户端调用API (示例使用 'mcp_use' 客户端): 配置好MCP客户端的JSON文件后，您的MCP客户端（例如，一个LLM代理）就可以发现并调用 'get_user' 这个工具了。

例如，通过 'mcp_use' 库（一个MCP客户端库）：

from mcp_use import MCPClient
import asyncio
import os
from pathlib import Path

async def main():
    # 假设 myapp.py 位于当前运行脚本的同级目录
    app_path = Path(__file__).parent / "myapp.py"
    if not app_path.exists():
        print(f"错误: 找不到应用程序文件 '{app_path}'。请确保文件存在或修改路径。")
        return

    config = {
        "mcpServers": {
            "My Simple API": {
                "command": "python",
                "args": [str(app_path)],
                "description": "一个提供用户数据的简单API。"
            }
        }
    }
    client = MCPClient(config=config)
    session = await client.create_session("My Simple API")

    print("\n调用 'get_user' 工具获取用户ID为1的信息...")
    result = await session.connector.call_tool("get_user", {"user_id": 1})
    print(f"返回结果: {result.content[0].text}")
    # 预期输出示例: {"id": 1, "name": "Alice Smith", "email": "[email protected]", "created_at": "2023-01-10T00:00:00Z"}

    print("\n调用 'get_user' 工具获取用户ID为999的信息 (不存在)...")
    result_not_found = await session.connector.call_tool("get_user", {"user_id": 999})
    print(f"返回结果: {result_not_found.content[0].text}")
    # 预期输出示例: {"id": 0, "name": "Not Found", "email": "[email protected]", "created_at": "1970-01-01T00:00:00Z"}


    await client.close_all_sessions()
    print("\n客户端会话已关闭。")

if __name__ == "__main__":
    asyncio.run(main())

要运行上述客户端代码，请将其保存为 'client_example.py' 并在 'myapp.py' 同级目录运行 'python client_example.py'。

信息

分类

开发者工具

访问项目