automcp

automcp

🚀 概述

automcp 可让你轻松地将现有 Agent 框架中的工具、Agent 和编排器转换为 MCP 服务，然后通过 Cursor 和 Claude Desktop 等客户端，使用标准化接口访问这些服务。

目前，我们支持将以下 Agent 框架中的 Agent、工具和编排器部署为 MCP 服务：

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

🔧 安装

从 PyPI 安装：

# Basic installation
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 服务：

进入实现了 Agent 的项目目录：

cd your-project-directory

使用以下标志之一（crewai、langgraph、llamaindex、openai、pydantic、mcp_agent）通过 CLI 生成 MCP 服务文件：

automcp init -f crewai

编辑生成的 run_mcp.py 文件以配置你的 Agent：

# Replace these imports with your actual agent classes
from your_module import YourCrewClass

# Define the input schema
class InputSchema(BaseModel):
    parameter1: str
    parameter2: str

# Set your agent details
name = "<YOUR_AGENT_NAME>"
description = "<YOUR_AGENT_DESCRIPTION>"

# For CrewAI projects
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 传输处理程序
• Agent 实现的占位内容
• 用于屏蔽可能破坏 STDIO 协议的警告的实用工具

你需要编辑此文件来：

• 导入你的 Agent/crew 类
• 定义输入架构（Agent 接受的参数）
• 使用你的 Agent 配置适配器

🔍 示例

运行示例

代码库为每个受支持的框架都提供了示例：

# Clone the repository
git clone https://github.com/NapthaAI/automcp.git
cd automcp

# Install automcp in development mode
pip install -e .

# Navigate to an example directory
cd examples/crewai/marketing_agents

# Generate the MCP server files (use the appropriate framework)
automcp init -f crewai

# Edit the generated run_mcp.py file to import and configure the example agent
# (See the specific example's README for details)

# Add a .env file with necessary environmental variables

# Install dependencies and run
automcp serve -t sse

每个示例都遵循与常规项目相同的工作流：

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

CrewAI 示例

下面是 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 = "A crew that posts marketing posts to a social media platform"

# Create an adapter for 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
)

# Server entrypoints
def serve_sse():
    mcp.run(transport="sse")

def serve_stdio():
    # Redirect stderr to suppress warnings that bypass the filters
    import os
    import sys

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

    # Save the original stderr
    original_stderr = sys.stderr

    # Replace stderr with our null writer to prevent warnings from corrupting STDIO
    sys.stderr = NullWriter()

    # Set environment variable to ignore Python warnings
    os.environ["PYTHONWARNINGS"] = "ignore"

    try:
        mcp.run(transport="stdio")
    finally:
        # Restore stderr for normal operation
        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 服务

设置好文件后，可以使用以下任一方法运行服务：

# Using the automcp CLI
automcp serve -t stdio    # STDIO transport
automcp serve -t sse      # SSE transport

# Or run the Python file directly
python run_mcp.py       # STDIO transport
python run_mcp.py sse   # SSE transport

# Or with uv run (if configured in pyproject.toml)
uv run serve_stdio
uv run serve_sse

关于传输模式的说明：

• STDIO：无需手动运行服务，客户端（Cursor）会启动它
• SSE：需要分两步操作：
  1. 单独启动服务：python run_mcp.py sse 或 automcp 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 = ["."]

如果你的 Agent 位于代码库的子目录/包中：

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

如果 MCP 服务成功启动并运行在端口 8000 上，就说明一切就绪！

启动服务

1. 前往 labs.naptha.ai
2. 使用你的 github 账户登录
3. 从代码库列表中选择你编辑的代码库——我们会自动发现你的 github 代码库。
4. 添加环境变量，例如 OPENAI_API_KEY 等。
5. 点击 Launch。
6. 复制 SSE URL，并将其粘贴到 MCP 客户端中：

🔌 与 MCP 客户端搭配使用

Cursor

要集成到 Cursor IDE 中，请在项目根目录创建一个 .cursor 文件夹，并添加一个包含以下配置的 mcp.json 文件：

{
    "mcpServers": {
        "crew-name-stdio": {
            "type": "stdio",
            "command": "/absolute/path/to/your/.venv/bin/uv",
            "args": [
                "--directory",
                "/absolute/path/to/your/project_dir",
                "run",
                "serve_stdio"
            ],
            "env": {
                "OPENAI_API_KEY": "sk-",
                "SERPER_API_KEY": ""
            }
        },
        
        "crew-name-python": {
            "type": "stdio",
            "command": "/absolute/path/to/your/.venv/bin/python",
            "args": [
                "/absolute/path/to/your/project_dir/run_mcp.py"
            ],
            "env": {
                "OPENAI_API_KEY": "sk-",
                "SERPER_API_KEY": ""
            }
        },
        
        "crew-name-automcp": {
            "type": "stdio",
            "command": "/absolute/path/to/your/.venv/bin/automcp",
            "args": [
                "serve",
                "-t",
                "stdio"
            ],
            "cwd": "/absolute/path/to/your/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"
           }
       }
   }
}

🛠️ 创建新适配器

想要为新的 Agent 框架添加支持？操作如下：

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:
    """Doc string for your function"""
    
    # Get the field names and types from the input schema
    schema_fields = input_schema.model_fields

    # Create the parameter string for the function signature
    params_str = ", ".join(
        f"{field_name}: {field_info.annotation.__name__}"
        for field_name, field_info in schema_fields.items()
    )

    # Create the function body that constructs the input schema
    # Note: You may need to adjust the method calls (kickoff, model_dump_json)
    # to match your framework's specific 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()
    """

    # Create a namespace for the function
    namespace = {
        "input_schema": input_schema,
        "agent_instance": agent_instance,
        "json": json,
        "contextlib": contextlib,
        "io": io,
    }

    # Execute the function definition in the namespace
    exec(body_str, namespace)

    # Get the created function
    run_agent = namespace["run_agent"]

    # Add proper function metadata
    run_agent.__name__ = name
    run_agent.__doc__ = description

    return run_agent

2. 在 examples/your_framework/ 中创建一个示例

📝 注意事项

• 使用 STDIO 传输时，请谨慎处理 Agent 代码中的 print 语句，因为它们可能会破坏协议
• 可使用 MCP Inspector 进行调试：npx @modelcontextprotocol/inspector
• 请记住，在 STDIO 模式下，客户端（如 Cursor）会为你启动服务
• 在 SSE 模式下，你需要手动启动服务，然后配置客户端连接到该服务