返回市场
自动-MCP

自动-MCP

作者:NapthaAI299 星标更新:2025-04-18

项目介绍

automcp

🚀 概述

automcp 允许您轻松地将现有代理框架中的工具、代理和编排器转换为 MCP 服务器,这些服务器可以通过客户端(如 Cursor 和 Claude Desktop)的标准接口进行访问。

我们目前支持以下代理框架的代理、工具和编排器作为 MCP 服务器的部署:

  1. CrewAI
  2. LangGraph
  3. Llama Index
  4. OpenAI Agents SDK
  5. Pydantic AI
  6. mcp-agent

🔧 安装

从 PyPI 安装:

# 基本安装
pip install naptha-automcp

# UV
uv add naptha-automcp

或者从源代码安装:

git clone https://github.com/napthaai/automcp.git
cd automcp
uv venv 
source .venv/bin/activate
pip install -e .

🧩 快速开始

为您的项目创建一个新的 MCP 服务器:

导航到包含您的代理实现的项目目录:

cd your-project-directory

通过 CLI 使用以下标志之一生成 MCP 服务器文件(crewai, langgraph, llamaindex, openai, pydantic, mcp_agent):

automcp init -f crewai

编辑生成的 run_mcp.py 文件以配置您的代理:

# 替换这些导入语句为您实际的代理类
from your_module import YourCrewClass

# 定义输入模式
class InputSchema(BaseModel):
    parameter1: str
    parameter2: str

# 设置您的代理详细信息
name = "<YOUR_AGENT_NAME>"
description = "<YOUR_AGENT_DESCRIPTION>"

# 对于 CrewAI 项目
mcp_crewai = create_crewai_adapter(
    orchestrator_instance=YourCrewClass().crew(),
    name=name,
    description=description,
    input_schema=InputSchema,
)

安装依赖项并运行您的 MCP 服务器:

automcp serve -t sse

📁 生成的文件

当您运行 automcp init -f <FRAMEWORK> 时,会生成以下文件:

run_mcp.py

这是设置和运行您的 MCP 服务器的主要文件。它包含:

  • 服务器初始化代码
  • STDIO 和 SSE 传输处理器
  • 您的代理实现占位符
  • 抑制可能破坏 STDIO 协议的警告的实用工具

您需要编辑此文件以:

  • 导入您的代理/团队类
  • 定义您的输入模式(您的代理接受的参数)
  • 使用您的代理配置适配器

🔍 示例

运行示例

仓库中包括每个支持框架的示例:

# 克隆仓库
git clone https://github.com/NapthaAI/automcp.git
cd automcp

# 在开发模式下安装 automcp
pip install -e .

# 导航到一个示例目录
cd examples/crewai/marketing_agents

# 生成 MCP 服务器文件(使用适当的框架)
automcp init -f crewai

# 编辑生成的 run_mcp.py 文件以导入和配置示例代理
# (参见特定示例的 README 以获取详细信息)

# 添加包含必要环境变量的 .env 文件

# 安装依赖项并运行
automcp serve -t sse

每个示例遵循与常规项目相同的流程:

  1. 运行 automcp init -f <FRAMEWORK> 以生成服务器文件
  2. 编辑 run_mcp.py 以导入和配置示例代理
  3. 添加包含必要环境变量的 .env 文件
  4. 安装依赖项并使用 automcp serve -t sse 提供服务

CrewAI 示例

这是一个典型的配置好的 run_mcp.py 文件示例:

import warnings
from typing import Any
from automcp.adapters.crewai import create_crewai_adapter
from pydantic import BaseModel
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MCP Server")

warnings.filterwarnings("ignore")

from crew import MarketingPostsCrew

class InputSchema(BaseModel):
    project_description: str
    customer_domain: str

name = "marketing_posts_crew"
description = "一个在社交媒体平台上发布营销帖子的团队"

# 创建一个 crewai 的适配器
mcp_crewai = create_crewai_adapter(
    orchestrator_instance=MarketingPostsCrew().crew(),
    name=name,
    description=description,
    input_schema=InputSchema,
)
mcp.add_tool(
    mcp_crewai,
    name=name,
    description=description
)

# 服务器入口点
def serve_sse():
    mcp.run(transport="sse")

def serve_stdio():
    # 将 stderr 重定向以抑制绕过过滤器的警告
    import os
    import sys

    class NullWriter:
        def write(self, *args, **kwargs):
            pass
        def flush(self, *args, **kwargs):
            pass

    # 保存原始 stderr
    original_stderr = sys.stderr

    # 用我们的 null writer 替换 stderr 以防止警告破坏 STDIO
    sys.stderr = NullWriter()

    # 设置环境变量以忽略 Python 警告
    os.environ["PYTHONWARNINGS"] = "ignore"

    try:
        mcp.run(transport="stdio")
    finally:
        # 恢复 stderr 以恢复正常操作
        sys.stderr = original_stderr

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "sse":
        serve_sse()
    else:
        serve_stdio()

🔄 运行您的 MCP 服务器

设置好文件后,您可以使用以下方法之一运行您的服务器:

# 使用 automcp CLI
automcp serve -t stdio    # STDIO 传输
automcp serve -t sse      # SSE 传输

# 或直接运行 Python 文件
python run_mcp.py       # STDIO 传输
python run_mcp.py sse   # SSE 传输

# 或使用 uv run(如果在 pyproject.toml 中配置)
uv run serve_stdio
uv run serve_sse

关于传输模式的注意事项:

  • STDIO:您不需要手动启动服务器——它将由客户端(Cursor)启动
  • SSE:这是一个两步过程:
    1. 分别启动服务器:python run_mcp.py sseautomcp serve -t sse
    2. 添加 mcp.json 配置以连接到正在运行的服务器

如果您想使用 uv run 命令,请在 pyproject.toml 中添加以下内容:

[tool.uv.scripts]
serve_stdio = "python run_mcp.py"
serve_sse = "python run_mcp.py sse"

☁️ 使用 Naptha 的 MCPaaS 部署

Naptha 支持将您新创建的 MCP 服务器部署到我们的 MCP 服务器即服务平台!开始使用非常简单。

设置

Naptha 的 MCPaaS 平台要求您的仓库已使用 uv 设置。 这意味着您需要在 pyproject.toml 中进行一些配置。

首先,确保由 naptha-automcp 生成的 run_mcp.py 文件是您仓库的根目录。

其次,确保您的 pyproject.toml 包含以下配置:

[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"

[project.scripts]
serve_stdio = "run_mcp:serve_stdio"
serve_sse = "run_mcp:serve_sse"

[tool.hatch.metadata]
allow-direct-references = true

[tool.hatch.build.targets.wheel]
include = [ "run_mcp.py",]
exclude = [ "__pycache__", "*.pyc",]
sources = [ ".",]
packages = ["."]

如果您的代理位于仓库的子目录或包中:

pyproject.toml
run_mcp.py
my_agent/
|---| __init__.py
    | agent.py

确保在 run_mcp.py 中像这样导入它:

from my_agent.agent

而不是如下所示,因为这会导致构建失败:

from .my_agent.agent

一旦配置完成,提交并推送到 GitHub(但不要推送环境变量!)。然后,您可以测试以确保一切设置正确:

uvx --from https://github.com/your-username/your-repo serve_sse

如果这成功地在端口 8000 上启动了您的 MCP 服务器,那么您就可以继续了!

启动您的服务器

  1. 访问 labs.naptha.ai
  2. 使用您的 GitHub 账户登录
  3. 从您的仓库列表中选择您编辑过的仓库——我们会自动发现您的 GitHub 仓库。
  4. 添加您的环境变量,例如 OPENAI_API_KEY 等。
  5. 点击启动。
  6. 复制 SSE URL,并将其粘贴到您的 MCP 客户端中:

🔌 使用 MCP 客户端

Cursor

要与 Cursor IDE 集成,在项目根目录中创建一个 .cursor 文件夹,并添加一个带有以下配置的 mcp.json 文件:

{
    "mcpServers": {
        "crew-name-stdio": {
            "type": "stdio",
            "command": "/绝对路径/到你的/.venv/bin/uv",
            "args": [
                "--directory",
                "/绝对路径/到你的/project_dir",
                "run",
                "serve_stdio"
            ],
            "env": {
                "OPENAI_API_KEY": "sk-",
                "SERPER_API_KEY": ""
            }
        },
        
        "crew-name-python": {
            "type": "stdio",
            "command": "/绝对路径/到你的/.venv/bin/python",
            "args": [
                "/绝对路径/到你的/project_dir/run_mcp.py"
            ],
            "env": {
                "OPENAI_API_KEY": "sk-",
                "SERPER_API_KEY": ""
            }
        },
        
        "crew-name-automcp": {
            "type": "stdio",
            "command": "/绝对路径/到你的/.venv/bin/automcp",
            "args": [
                "serve",
                "-t",
                "stdio"
            ],
            "cwd": "/绝对路径/到你的/project_dir",
            "env": {
                "OPENAI_API_KEY": "sk-",
                "SERPER_API_KEY": ""
            }
        },
        
        "crew-name-sse": {
            "type": "sse",
            "url": "http://localhost:8000/sse"
        }
    }
}

注意: 确保替换所有占位符路径为实际文件和目录的绝对路径。

直接 GitHub 执行

将您的项目推送到 GitHub 并使用:

{
   "mcpServers": {
       "My Agent": {
           "command": "uvx",
           "args": [
               "--from",
               "git+https://github.com/your-username/your-repo",
               "serve_stdio"
           ],
           "env": {
               "OPENAI_API_KEY": "your-key-here"
           }
       }
   }
}

🛠️ 创建新的适配器

想要添加对新代理框架的支持?这里是如何操作的:

  1. 在 automcp/adapters/ 中创建一个新的适配器文件(或添加到现有的框架文件中):
# automcp/adapters/framework.py
import json
import contextlib
import io
from typing import Any, Callable, Type
from pydantic import BaseModel

def create_framework_adapter(
    agent_instance: Any,
    name: str,
    description: str,
    input_schema: Type[BaseModel],
) -> Callable:
    """您的函数的文档字符串"""
    
    # 从输入模式中获取字段名称和类型
    schema_fields = input_schema.model_fields

    # 创建函数签名的参数字符串
    params_str = ", ".join(
        f"{field_name}: {field_info.annotation.__name__}"
        for field_name, field_info in schema_fields.items()
    )

    # 创建构造输入模式的函数主体
    # 注意:您可能需要调整方法调用(kickoff, model_dump_json)
    # 以匹配您框架的具体 API
    body_str = f"""def run_agent({params_str}):
        inputs = input_schema({', '.join(f'{name}={name}' for name in schema_fields)})
        with contextlib.redirect_stdout(io.StringIO()):
            result = agent_instance.framework_specific_run(inputs=inputs.model_dump())
        return result.framework_specific_result()
    """

    # 为函数创建命名空间
    namespace = {
        "input_schema": input_schema,
        "agent_instance": agent_instance,
        "json": json,
        "contextlib": contextlib,
        "io": io,
    }

    # 在命名空间中执行函数定义
    exec(body_str, namespace)

    # 获取创建的函数
    run_agent = namespace["run_agent"]

    # 添加正确的函数元数据
    run_agent.__name__ = name
    run_agent.__doc__ = description

    return run_agent
  1. 在 examples/your_framework/ 中创建一个示例

📝 注意事项

  • 当使用 STDIO 传输时,小心不要在代理代码中使用 print 语句,因为它们可能会破坏协议
  • 可以使用 MCP Inspector 进行调试:npx @modelcontextprotocol/inspector
  • 记住,对于 STDIO 模式,客户端(如 Cursor)会为您启动服务器
  • 对于 SSE 模式,您需要手动启动服务器,然后配置客户端连接到它