automcp
🚀 概述
automcp允许您轻松地将工具、代理和编排器从现有的代理框架转换为 主控程序 服务器，然后可以通过Cursor和Claude Desktop等客户端通过标准化接口访问。
我们目前支持将代理、工具和编排器部署为以下代理框架的MCP服务器：
- CrewAI
- LangGraph
- 火焰指数
- OpenAI代理SDK
- Pydantic 人工智能
- mcp代理
🔧 安装
从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服务器：
使用代理实现导航到项目目录：
cd your-project-directory使用以下标志之一（creuai、langgraph、llamaindex、openai、pydantic、MCP_agent）通过CLI生成MCP服务器文件：
automcp init -f crewai编辑生成的 run_mcp.py 配置代理的文件：
# 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 = ""
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 ，生成以下文件：
run_mcp.py
这是设置和运行MCP服务器的主文件。它包含：
- 服务器初始化代码
- STDIO和SSE传输处理程序
- 代理实现的占位符
- 用于抑制可能损坏STDIO协议的警告的实用程序
您需要将此文件编辑为：
- 导入您的代理/船员课程
- 定义您的输入模式（您的代理接受的参数）
- 使用代理配置适配器
🔍 例子
运行示例
存储库包括每个受支持框架的示例：
# 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每个示例都遵循与常规项目相同的工作流程：
- 跑
automcp init -f生成服务器文件 - 编辑
run_mcp.py导入和配置示例代理 - 添加一个包含必要环境变量的.env文件
- 安装依赖项并使用
automcp serve -t sse
CrewAI示例
以下是典型的配置 run_mcp.py 以CrewAI为例：
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关于运输方式的说明：
- 标准输入输出：您不需要手动运行服务器，它将由客户端（Cursor）启动
- SSE：这是一个两步过程：
1. 单独启动服务器： python run_mcp.py sse 或 automcp serve -t sse 1. 添加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.
首先，确保 run_mcp.py 由生成的文件 naptha-automcp 是存储库的根。
第二，确保你的 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如果这导致您的MCP服务器在端口8000上成功启动，那么您就可以开始了！
启动服务器
- 首选 拿普塔实验室
- 使用您的github帐户登录
- 从存储库列表中选择您编辑的存储库——我们会自动发现您的github存储库。
- 添加您的环境变量，例如。
OPENAI_API_KEY等等。 - 单击启动。
- 复制SSE URL，并将其粘贴到MCP客户端：
🔌 与MCP客户端一起使用
光标
要与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"
}
}
}
}🛠️ 创建新适配器
想要添加对新代理框架的支持吗？方法如下：
- 在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- 在examples/your_framework中创建一个示例/
📝 备注
- 使用STDIO传输时，请小心代理代码中的print语句，因为它们可能会破坏协议
- MCP检查器可用于调试：
npx @modelcontextprotocol/inspector - 请记住，对于STDIO模式，客户端（如Cursor）将为您启动服务器
- 对于SSE模式，您需要手动启动服务器，然后配置客户端以连接到它