MCP代理构建工具

MCP-Use是一个开源工具,用于连接任何LLM到MCP服务器并构建具有工具访问权限的自定义MCP代理。

  • by marcusRB
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 marcusRB/mcp-model-context-protocol

使用说明

🚀 Create MCP Clients and Agents

🌐 MCP-Use 是连接的开源方式 任何大型语言模型(LLM)到任何MCP服务器 并构建具有工具访问权限的自定义MCP代理,而无需使用闭源或应用程序客户端。

💡 让开发者能够轻松地将任何大型语言模型(LLM)连接到网络浏览、文件操作等工具。

支持的功能 (此处可填写具体支持的功能或特性,由于原文未提供具体内容,故留空)
基本元素/原始元素
交通

特点/功能

Feature Description
🔄 Ease of use Create your first MCP capable agent you need only 6 lines of code
🤖 LLM Flexibility Works with any langchain supported LLM that supports tool calling (OpenAI, Anthropic, Groq, LLama etc.)
🌐 Code Builder Explore MCP capabilities and generate starter code with the interactive code builder.
🔗 HTTP Support Direct connection to MCP servers running on specific HTTP ports
⚙️ Dynamic Server Selection Agents can dynamically choose the most appropriate MCP server for a given task from the available pool
🧩 Multi-Server Support Use multiple MCP servers simultaneously in a single agent
🛡️ Tool Restrictions Restrict potentially dangerous tools like file system or network access
🔧 Custom Agents Build your own agents with any framework using the LangChain adapter or create new adapters
❓ What should we build next Let us know what you'd like us to build next

快速入门

使用 pip:

pip install mcp-use

或者从源代码安装:

git clone https://github.com/mcp-use/mcp-use.git
cd mcp-use
pip install -e .

安装LangChain提供者

mcp_use 通过 LangChain 与各种大型语言模型(LLM)提供商合作。您需要为所选的大型语言模型安装相应的 LangChain 提供商包。例如:

# For OpenAI
pip install langchain-openai

# For Anthropic
pip install langchain-anthropic

对于其他提供商,请检查 LangChain聊天模型文档 并添加您想要使用的提供商的API密钥到您的 .env 文件。

OPENAI_API_KEY=
ANTHROPIC_API_KEY=

重要的只有具备工具调用能力的模型才能与mcp_use一起使用。请确保您选择的模型支持函数调用或工具使用。

启动您的代理:

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient

async def main():
    # Load environment variables
    load_dotenv()

    # Create configuration dictionary
    config = {
      "mcpServers": {
        "playwright": {
          "command": "npx",
          "args": ["@playwright/mcp@latest"],
          "env": {
            "DISPLAY": ":1"
          }
        }
      }
    }

    # Create MCPClient from configuration dictionary
    client = MCPClient.from_dict(config)

    # Create LLM
    llm = ChatOpenAI(model="gpt-4o")

    # Create agent with the client
    agent = MCPAgent(llm=llm, client=client, max_steps=30)

    # Run the query
    result = await agent.run(
        "Find the best restaurant in San Francisco",
    )
    print(f"\nResult: {result}")

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

你也可以从配置文件中添加服务器配置,如下所示:

client = MCPClient.from_config_file(
        os.path.join("browser_mcp.json")
    )

示例配置文件(browser_mcp.json):

{
  "mcpServers": {
    "playwright": {
      "command": "npx",
      "args": ["@playwright/mcp@latest"],
      "env": {
        "DISPLAY": ":1"
      }
    }
  }
}

如需了解其他设置、模型及更多信息,请查阅文档。

流式代理输出

MCP-Use 支持使用异步流处理代理输出 astream 方法(或“途径”)在于 MCPAgent这允许您实时接收由代理生成的增量结果、工具操作以及中间步骤,从而实现即时反馈和进度报告。

如何使用

呼叫 agent.astream(query) 并异步迭代处理结果:

async for chunk in agent.astream("Find the best restaurant in San Francisco"):
    print(chunk["messages"], end="", flush=True)

每个数据块都是一个字典,包含诸如以下键: actions, steps, messages,并且(在最后一部分) output这使您能够构建响应式用户界面或实时记录代理进度。

示例:实际应用中的流媒体

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient

async def main():
    load_dotenv()
    client = MCPClient.from_config_file("browser_mcp.json")
    llm = ChatOpenAI(model="gpt-4o")
    agent = MCPAgent(llm=llm, client=client, max_steps=30)
    async for chunk in agent.astream("Look for job at nvidia for machine learning engineer."):
        print(chunk["messages"], end="", flush=True)

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

这个流式接口非常适合需要实时更新的应用程序,例如聊天机器人、仪表板或交互式笔记本。

示例用例

使用 Playwright 进行网页浏览

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient

async def main():
    # Load environment variables
    load_dotenv()

    # Create MCPClient from config file
    client = MCPClient.from_config_file(
        os.path.join(os.path.dirname(__file__), "browser_mcp.json")
    )

    # Create LLM
    llm = ChatOpenAI(model="gpt-4o")
    # Alternative models:
    # llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
    # llm = ChatGroq(model="llama3-8b-8192")

    # Create agent with the client
    agent = MCPAgent(llm=llm, client=client, max_steps=30)

    # Run the query
    result = await agent.run(
        "Find the best restaurant in San Francisco USING GOOGLE SEARCH",
        max_steps=30,
    )
    print(f"\nResult: {result}")

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

Airbnb搜索

import asyncio
import os
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from mcp_use import MCPAgent, MCPClient

async def run_airbnb_example():
    # Load environment variables
    load_dotenv()

    # Create MCPClient with Airbnb configuration
    client = MCPClient.from_config_file(
        os.path.join(os.path.dirname(__file__), "airbnb_mcp.json")
    )

    # Create LLM - you can choose between different models
    llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

    # Create agent with the client
    agent = MCPAgent(llm=llm, client=client, max_steps=30)

    try:
        # Run a query to search for accommodations
        result = await agent.run(
            "Find me a nice place to stay in Barcelona for 2 adults "
            "for a week in August. I prefer places with a pool and "
            "good reviews. Show me the top 3 options.",
            max_steps=30,
        )
        print(f"\nResult: {result}")
    finally:
        # Ensure we clean up resources properly
        if client.sessions:
            await client.close_all_sessions()

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

示例配置文件(airbnb_mcp.json):

{
  "mcpServers": {
    "airbnb": {
      "command": "npx",
      "args": ["-y", "@openbnb/mcp-server-airbnb"]
    }
  }
}

Blender 3D创作

import asyncio
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from mcp_use import MCPAgent, MCPClient

async def run_blender_example():
    # Load environment variables
    load_dotenv()

    # Create MCPClient with Blender MCP configuration
    config = {"mcpServers": {"blender": {"command": "uvx", "args": ["blender-mcp"]}}}
    client = MCPClient.from_dict(config)

    # Create LLM
    llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

    # Create agent with the client
    agent = MCPAgent(llm=llm, client=client, max_steps=30)

    try:
        # Run the query
        result = await agent.run(
            "Create an inflatable cube with soft material and a plane as ground.",
            max_steps=30,
        )
        print(f"\nResult: {result}")
    finally:
        # Ensure we clean up resources properly
        if client.sessions:
            await client.close_all_sessions()

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

配置支持

HTTP 连接示例

MCP-Use支持HTTP连接,使您能够连接到运行在特定HTTP端口上的MCP服务器。此功能对于与基于Web的MCP服务器进行集成特别有用。

以下是如何使用HTTP连接功能的一个示例:

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient

async def main():
    """Run the example using a configuration file."""
    # Load environment variables
    load_dotenv()

    config = {
        "mcpServers": {
            "http": {
                "url": "http://localhost:8931/sse"
            }
        }
    }

    # Create MCPClient from config file
    client = MCPClient.from_dict(config)

    # Create LLM
    llm = ChatOpenAI(model="gpt-4o")

    # Create agent with the client
    agent = MCPAgent(llm=llm, client=client, max_steps=30)

    # Run the query
    result = await agent.run(
        "Find the best restaurant in San Francisco USING GOOGLE SEARCH",
        max_steps=30,
    )
    print(f"\nResult: {result}")

if __name__ == "__main__":
    # Run the appropriate example
    asyncio.run(main())

此示例展示了如何连接到运行在特定HTTP端口上的MCP服务器。请确保在运行此示例之前已启动您的MCP服务器。

多服务器支持

MCP-Use 允许同时配置和连接到多个 MCP 服务器,使用的是 MCPClient这使得能够执行复杂的操作流程,这些流程需要从不同服务器获取工具,例如将网页浏览与文件操作或3D建模相结合。

配置

您可以在配置文件中设置多个服务器:

{
  "mcpServers": {
    "airbnb": {
      "command": "npx",
      "args": ["-y", "@openbnb/mcp-server-airbnb", "--ignore-robots-txt"]
    },
    "playwright": {
      "command": "npx",
      "args": ["@playwright/mcp@latest"],
      "env": {
        "DISPLAY": ":1"
      }
    }
  }
}

用法

这个(或:该) MCPClient 该类提供了管理与多个服务器连接的方法。在创建一个 MCPAgent,你可以提供一个 MCPClient 配置了多台服务器。

默认情况下,代理将能够访问所有已配置服务器上的工具。如果您需要针对特定任务指定一个特定的服务器,您可以指定该服务器 server_name 当调用时 agent.run() 方法。

# Example: Manually selecting a server for a specific task
result = await agent.run(
    "Search for Airbnb listings in Barcelona",
    server_name="airbnb" # Explicitly use the airbnb server
)

result_google = await agent.run(
    "Find restaurants near the first result using Google Search",
    server_name="playwright" # Explicitly use the playwright server
)

动态服务器选择(服务器管理器)

为了提高效率并减少在处理来自不同服务器的多个工具时可能产生的代理混淆,您可以通过设置来启用服务器管理器 use_server_manager=True 在……期间 MCPAgent 初始化。

当启用时,代理会根据大型语言模型(LLM)为特定步骤选择的工具,智能地选择正确的MCP服务器。这减少了不必要的连接,并确保代理为任务使用合适的工具。

import asyncio
from mcp_use import MCPClient, MCPAgent
from langchain_anthropic import ChatAnthropic

async def main():
    # Create client with multiple servers
    client = MCPClient.from_config_file("multi_server_config.json")

    # Create agent with the client
    agent = MCPAgent(
        llm=ChatAnthropic(model="claude-3-5-sonnet-20240620"),
        client=client,
        use_server_manager=True  # Enable the Server Manager
    )

    try:
        # Run a query that uses tools from multiple servers
        result = await agent.run(
            "Search for a nice place to stay in Barcelona on Airbnb, "
            "then use Google to find nearby restaurants and attractions."
        )
        print(result)
    finally:
        # Clean up all sessions
        await client.close_all_sessions()

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

工具访问控制

MCP-Use 允许您限制代理可用的工具,从而更好地保障安全并控制代理的功能:

import asyncio
from mcp_use import MCPAgent, MCPClient
from langchain_openai import ChatOpenAI

async def main():
    # Create client
    client = MCPClient.from_config_file("config.json")

    # Create agent with restricted tools
    agent = MCPAgent(
        llm=ChatOpenAI(model="gpt-4"),
        client=client,
        disallowed_tools=["file_system", "network"]  # Restrict potentially dangerous tools
    )

    # Run a query with restricted tool access
    result = await agent.run(
        "Find the best restaurant in San Francisco"
    )
    print(result)

    # Clean up
    await client.close_all_sessions()

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

沙盒执行

MCP-Use 支持在沙盒环境中使用 E2B 的云基础设施运行 MCP 服务器。这样,您无需在本地安装依赖项即可运行 MCP 服务器,使得使用那些可能具有复杂设置或系统要求的工具变得更加容易。

安装

要使用沙盒执行,您需要安装E2B依赖项:

# Install mcp-use with E2B support
pip install "mcp-use[e2b]"

# Or install the dependency directly
pip install e2b-code-interpreter

您还需要一个E2B API密钥。您可以在以下网址注册: e2b.dev 获取您的API密钥。

配置

要启用沙盒执行,请在创建时使用沙盒参数 MCPClient:

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient
from mcp_use.types.sandbox import SandboxOptions

async def main():
    # Load environment variables (needs E2B_API_KEY)
    load_dotenv()

    # Define MCP server configuration
    server_config = {
        "mcpServers": {
            "everything": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-everything"],
            }
        }
    }

    # Define sandbox options
    sandbox_options: SandboxOptions = {
        "api_key": os.getenv("E2B_API_KEY"),  # API key can also be provided directly
        "sandbox_template_id": "base",  # Use base template
    }

    # Create client with sandboxed mode enabled
    client = MCPClient(
        config=server_config,
        sandbox=True,
        sandbox_options=sandbox_options,

    )

    # Create agent with the sandboxed client
    llm = ChatOpenAI(model="gpt-4o")
    agent = MCPAgent(llm=llm, client=client)

    # Run your agent
    result = await agent.run("Use the command line tools to help me add 1+1")
    print(result)

    # Clean up
    await client.close_all_sessions()

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

沙盒选项

这个(或“该”) SandboxOptions 该类型为沙盒环境提供配置:

选项 描述 默认值
api_key E2B API密钥。必填项-可直接提供或通过E2B_API_KEY环境变量提供 无
sandbox_template_id 沙箱环境的模板ID "base"
supergateway_command 运行 supergateway 的命令 "npx -y supergateway"

沙盒执行的优势

  • 无本地依赖在不安装本地依赖项的情况下运行MCP服务器
  • 隔离在安全、隔离的环境中执行代码
  • 一致的环境确保在不同系统间的行为一致性
  • 资源效率将资源密集型任务卸载到云基础设施

直接工具调用(无需大型语言模型)

当你需要程序化控制时,可以直接调用MCP服务器工具,而无需使用LLM:

import asyncio
from mcp_use import MCPClient

async def call_tool_example():
    config = {
        "mcpServers": {
            "everything": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-everything"],
            }
        }
    }

    client = MCPClient.from_dict(config)

    try:
        await client.create_all_sessions()
        session = client.get_session("everything")

        # Call tool directly
        result = await session.call_tool(
            name="add",
            arguments={"a": 1, "b": 2}
        )

        print(f"Result: {result.content[0].text}")  # Output: 3

    finally:
        await client.close_all_sessions()

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

查看完整示例: 示例/直接工具调用.py

构建自定义代理:

你也可以使用LangChain适配器来构建你自己的自定义代理:

import asyncio
from langchain_openai import ChatOpenAI
from mcp_use.client import MCPClient
from mcp_use.adapters.langchain_adapter import LangChainAdapter
from dotenv import load_dotenv

load_dotenv()


async def main():
    # Initialize MCP client
    client = MCPClient.from_config_file("examples/browser_mcp.json")
    llm = ChatOpenAI(model="gpt-4o")

    # Create adapter instance
    adapter = LangChainAdapter()
    # Get LangChain tools with a single line
    tools = await adapter.create_tools(client)

    # Create a custom LangChain agent
    llm_with_tools = llm.bind_tools(tools)
    result = await llm_with_tools.ainvoke("What tools do you have available ? ")
    print(result)


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


调试

MCP-Use 提供了一个内置的调试模式,该模式会增加日志的详细程度,有助于诊断代理实现中的问题。

启用调试模式

启用调试模式主要有两种方式:

1. 环境变量(建议用于一次性运行)

使用以下方式运行你的脚本 DEBUG 环境变量设置为所需的级别:

# Level 1: Show INFO level messages
DEBUG=1 python3.11 examples/browser_use.py

# Level 2: Show DEBUG level messages (full verbose output)
DEBUG=2 python3.11 examples/browser_use.py

这仅在特定的Python进程运行期间设置调试级别。

或者,您可以将以下环境变量设置为您所需的日志记录级别:

export MCP_USE_DEBUG=1 # or 2

2. 通过编程设置调试标志

你可以在代码中直接设置全局调试标志:

import mcp_use

mcp_use.set_debug(1)  # INFO level
# or
mcp_use.set_debug(2)  # DEBUG level (full verbose output)

3. 代理特定的详细程度(或:针对代理的详尽性)

如果您只想查看代理的调试信息而不启用完整的调试日志记录,您可以设置 verbose 在创建 MCPAgent 时的参数:

# Create agent with increased verbosity
agent = MCPAgent(
    llm=your_llm,
    client=your_client,
    verbose=True  # Only shows debug messages from the agent
)

当你只需要查看代理的步骤和决策过程,而不需要其他组件的所有低级调试信息时,这会非常有用。

明星历史

做出贡献

我们热爱贡献!请随时提出问题以报告错误或功能请求。请查看 CONTRIBUTING.md 翻译为中文是:“贡献指南.md” 或 “如何贡献.md”(具体翻译可能根据上下文有所调整,但基本意思是关于如何为项目做出贡献的指南文件) 作为指导方针。

贡献者

感谢我们所有杰出的贡献者!

要求

  • Python 3.11及以上版本
  • MCP 实施(如 Playwright MCP)
  • LangChain 及相应的模型库(如 OpenAI、Anthropic 等)

许可证

麻省理工学院(MIT)

引文

这个MCP仓库的作者是:

@software{mcp_use2025,
  author = {Zullo, Pietro},
  title = {MCP-Use: MCP Library for Python},
  year = {2025},
  publisher = {GitHub},
  url = {https://github.com/pietrozullo/mcp-use}
}