我们正站在新一代软件的门槛上。在 YC Startup School 的最新主题演讲中，Andrej Karpathy 描述了 Software 3.0 的世界：自然语言成为主要编程接口，大型语言模型既是计算引擎，也是自主协作者。
与 Software 1.0（开发者编写明确的指令）或 Software 2.0（神经网络从数据中学习模式）不同，Software 3.0 运行在自然语言之上。我们的提示变成了程序，而 LLM 则将我们的意图转化为可执行的行为。
在本教程中，我们将构建一个 MCP 服务器，提供对 Novita AI 沙箱（sandbox）的远程访问，让 AI 代理可以安全地执行代码。该代理本身将使用 mcp-use 库创建，该库还能自动管理与 MCP 服务器的通信。
什么是 MCP（模型上下文协议）？
模型上下文协议（MCP）是由 Anthropic 开发的一个开放标准，用于 AI 模型与外部服务、工具和数据源进行通信。你可以把它想象成“AI 界的 USB-C”——它为代理与外部世界之间提供了标准接口。
三个核心组件：
- 工具（Tools）： 服务器暴露的可调用函数/API（例如，网页浏览、代码执行）。
- 资源（Resources）： 服务器提供的外部数据，通常作为 AI 代理的上下文（例如，文件、元数据、数据集）。
- 提示（Prompts）： 详细指令，用于指导代理在使用上述工具或资源时的行为。
关于 MCP 的架构和组件的深入介绍，请查看我们的博客：构建你的第一个 MCP 服务器。
为什么 MCP 对 AI 工具集成如此重要
MCP 建立了 AI 代理与外部系统之间的双向通信。这使得将 LLM“插入”自定义工具变得简单，同时减少了集成工作并降低了出错的可能性。
mcp-use 库简介
mcp-use 库是一个 Python 包，简化了与 MCP 服务器交互的 AI 代理的构建。它处理代理的创建，并管理与 MCP 服务器的通信以访问外部工具和数据源，让你只需专注于应用逻辑。
Novita AI 沙箱与模型 API 概述
什么是沙箱以及它为何重要
沙箱是一个安全的、隔离的运行时环境，可以在不影响主机系统的情况下执行不受信任的代码。本质上，它是一台轻量级的虚拟计算机，你的 AI 代理可以在其中执行代码、命令、创建文件等。
Novita AI 在云端提供这个沙箱，方便你的代理按需快速访问，并根据所用资源按秒灵活计费。
Novita 沙箱的主要特性：
- 安全隔离： 每个沙箱拥有独立的文件系统和环境，保护数据并防止意外交互。
- 快速启动： 沙箱实例平均启动时间低于约 200ms，非常适合低延迟场景。
- 多语言支持： 可以运行多种编程语言的代码，包括 Python、JavaScript、TypeScript 等。
- 快速暂停与恢复： 随时暂停沙箱，并在需要时恢复，文件系统和进程状态完全恢复。
- 后台执行： 支持后台任务执行，适用于需要等待结果的场景。
Novita 模型 API：
Novita 提供了来自 OpenAI、Google、DeepSeek、Qwen 等领先研究实验室的大量开源 AI 模型，涵盖语言、视觉、音频、视频和嵌入等领域。我们的语言模型与 OpenAI SDK 完全兼容，因此从 OpenAI 切换到 Novita 只需更新客户端的基础 URL 和 API 密钥，然后选择一个 Novita 模型即可。
| from openai import OpenAI client = OpenAI( base_url=“https://api.novita.ai/v3/openai”, api_key=“ ) |
设置开发环境
首先，克隆 GitHub 仓库，设置一个干净的 Python 环境，安装所有必需的依赖项，并获取 Novita AI 密钥。
使用 uv 克隆 GitHub 仓库并安装依赖项
1. 安装 uv（轻量级 Python 包管理器）
| pip install uv |
2. 克隆仓库（GitHub repo）并进入目录。
| git clone https://github.com/Studio1HQ/mcp\_remote\_execution.git cd mcp_remote_execution |
3. 创建并激活 uv 虚拟环境
| # 创建虚拟环境 uv venv # 激活虚拟环境 source .venv/bin/activate # Mac/Linux # 或 .venv\Scripts\activate # Windows |
4. 安装项目依赖
| # 安装依赖 uv sync |
创建 Novita AI 账户并获取 API 密钥
1. 在 novita.ai 注册。
2. 在控制台中，将鼠标悬停在用户头像图标上，点击弹出菜单中的 API Keys。
3. 在密钥管理页面，点击 Add New Key。在弹出的窗口中输入密钥名称，点击 Confirm，然后复制生成的密钥。
4. 在项目目录中创建 .env 文件，并粘贴以下内容。
| NOVITA_API_KEY=“<粘贴你的 Novita API 密钥>” NOVITA_BASE_URL=“https://api.novita.ai/v3/openai” NOVITA_E2B_DOMAIN=“sandbox.novita.ai” NOVITA_E2B_TEMPLATE=“code-interpreter-v1” |
为 Novita AI 账户充值积分
要使用 Novita 沙箱，你需要为账户充值积分。在控制台标签页中，点击 ‘Billing’。然后在账单页面，添加支付方式并至少充值 10 美元积分。
构建 MCP 服务器与 AI 代理集成
现在环境已设置好，让我们开始构建。这个过程分为两部分：首先，我们将逐步创建 MCP 服务器；然后，我们将构建 AI 代理应用程序（它充当 MCP 客户端）。但在开始之前，我们先创建一个 MCP 服务器将要使用的沙箱管理器。
沙箱管理器
沙箱管理器负责启动和停止沙箱实例。在我们的设置中，我们将服务器限制为一次只能有一个沙箱实例。沙箱管理器还将处理在沙箱中执行 Python 代码和运行 shell 命令。
首先，在 sandbox_manager.py 中，我们有 SandboxManager 类，它接受以下参数：
- sandbox_template：用于创建沙箱实例的模板。我们将使用 “code-interpreter-v1”，该模板预装了常用的 Python 包（例如 pandas、numpy）。
- sandbox_domain：用于连接 Novita AI 沙箱实例的域端点。
- sandbox_timeout：决定沙箱在被自动终止前保持活动状态的持续时间（以秒为单位）。
| from novita_sandbox.code_interpreter import Sandbox class SandboxManager: def __init__( self, sandbox_template: str, sandbox_domain: str, sandbox_timeout: int, ): self.sandbox_template = sandbox_template self.sandbox_domain = sandbox_domain self.sandbox_timeout = sandbox_timeout |
现在添加创建和停止沙箱的方法。
- 对于创建，我们接受 sandbox_api_key，使用它为用户启动一个新的沙箱实例，并返回带有沙箱 ID 的成功消息，或者如果出错则返回异常消息。
- 对于停止，我们接受 sandbox_api_key 和 sandbox_id，连接到沙箱，如果存在则停止它。与之前一样，根据结果返回成功消息或异常消息。
| … class SandboxManager: … # 以下现有代码 def create_sandbox_session(self, sandbox_api_key: str) -> str: “”“ 这将创建一个新的沙箱实例。 参数： sandbox_api_key (str): 沙箱的 API 密钥。 返回： str: 如果成功则返回带有新沙箱 ID 的消息，否则返回错误消息。 ”“” try: # 创建新的沙箱 sandbox = Sandbox.create( template=self.sandbox_template, api_key=sandbox_api_key, domain=self.sandbox_domain, timeout=self.sandbox_timeout, ) return f"成功创建沙箱。沙箱 ID: {sandbox.sandbox_id}“ except Exception as e: return f"创建新沙箱失败:{str(e)}” def stop_sandbox_session(self, sandbox_api_key: str, sandbox_id: str) -> str: “”“ 如果存在，这将销毁一个沙箱实例。 参数： sandbox_api_key (str): 沙箱的 API 密钥。 sandbox_id (str): 沙箱的 ID。 返回： str: 如果成功则返回带有被销毁沙箱 ID 的消息，否则返回错误消息。 ”“” try: # 连接到沙箱 sandbox = Sandbox.connect( api_key=sandbox_api_key, sandbox_id=sandbox_id, ) sandbox.kill() return f"成功销毁沙箱 ID: {sandbox_id}“ except Exception as e: return f"销毁沙箱 ID: {sandbox_id} 失败\ {str(e)}” |
最后，我们添加在用户沙箱中运行 Python 代码和 shell 命令的方法，通过 API 密钥和 ID 连接到沙箱。所有沙箱输出（包括异常和错误）都以字典形式返回。
| … class SandboxManager: … # 以下现有代码 def run_python_code( self, python_code: str, sandbox_api_key: str, sandbox_id: str ) -> dict: “”“ 在沙箱中运行 Python 代码，如果有图像输出则跳过。 参数： python_code (str): 要运行的 Python 代码。 sandbox_api_key (str): 沙箱的 API 密钥。 sandbox_id (str): 沙箱的 ID。 返回： dict: 包含 stdout、logs、error 等。 ”“” try: # 连接到沙箱 sandbox = Sandbox.connect( api_key=sandbox_api_key, sandbox_id=sandbox_id, ) execution = sandbox.run_code(python_code, language=“python”) return { # 跳过图像输出。 “outputs”: [result for result in execution.results if not result.png], “logs”: execution.logs, “error”: execution.error, } except Exception as e: return {“error”: str(e)} def run_on_command_line( self, command: str, sandbox_api_key: str, sandbox_id: str ) -> dict: “”“ 在沙箱中运行命令。 参数： command (str): 要运行的命令。 sandbox_api_key (str): 沙箱的 API 密钥。 sandbox_id (str): 沙箱的 ID。 返回： dict: 包含命令输出和执行错误（如有）。 ”“” try: # 连接到沙箱 sandbox = Sandbox.connect( api_key=sandbox_api_key, sandbox_id=sandbox_id, ) result = sandbox.commands.run(command) return { “output”: { “stdout”: result.stdout, “stderr”: result.stderr, “exit_code”: result.exit_code, “error”: result.error, }, “execution error”: None, } except Exception as e: return {“output”: None, “execution error”: str(e)} |
创建 MCP 服务器
现在沙箱管理器已设置好，我们开始处理 mcp_server.py。首先，创建一个 FastMCP 实例（这是运行服务器的框架），以及一个沙箱管理器。我们还创建一个 Rich 控制台实例，用于在终端中漂亮地打印输出。
注意： 传递给沙箱管理器的沙箱超时时间将用于此服务器上启动的每个沙箱。它是每个沙箱保持活动（运行）状态的最长时间（以秒为单位），除非提前停止。
| import asyncio import os from dotenv import load_dotenv from mcp.server.fastmcp import Context, FastMCP from rich.console import Console from rich.panel import Panel from rich.table import Table from starlette.requests import Request from sandbox_manager import SandboxManager # 加载 .env 变量 load_dotenv() console = Console() # 初始化 FastMCP 服务器 mcp = FastMCP(“MCP_Server”) # 为单例沙箱实例初始化沙箱管理器。 sandbox_manager = SandboxManager( sandbox_template=os.getenv(“NOVITA_E2B_TEMPLATE”), sandbox_domain=os.getenv(“NOVITA_E2B_DOMAIN”), sandbox_timeout=900, # 900 秒（15 分钟），沙箱实例将在之后自动销毁。 ) |
现在，由于需要 API 密钥才能连接到用户的沙箱，我们将从用户对服务器的请求中获取它。稍后我们会在 MCP 客户端中看到，这个密钥将在每次请求的授权头中发送。这意味着我们需要一种在服务器端从该头中提取 API 密钥的方法。
FastMCP 提供了一个 Context 对象，在运行时包含发出请求的用户的信息。因此，我们将创建一个辅助方法 get_user_api_key，它接受一个上下文对象，从头中提取 API 密钥并返回它，如果缺少则引发异常。
| … # 以下现有代码 def get_user_api_key(ctx: Context) -> str: “”“ 如果存在，从请求头中返回 API 密钥，否则引发异常。 ”“” request: Request = ctx.request_context.request # 访问请求数据 auth_header = request.headers.get(“Authorization”) if auth_header: auth_header = auth_header.split(" ")[1] if not auth_header: raise Exception(“Authorization Bearer 头中缺少 API 密钥”) return auth_header |
现在，在服务器上暴露 prompts、tools 和 resources。为此，我们只需在 Python 函数上添加装饰器 @mcp.{prompt, tool, resource}()。从我们的 prompt 开始，它返回关于代理应如何使用沙箱的指令。
| … # 以下现有代码（注意：为了方便，跳过了 Console 显示辅助方法） @mcp.prompt() def instructions_for_sandbox_use() -> str: “”“ 返回关于沙箱使用的必须阅读的指令。 ”“” return “”“ 当你想使用沙箱功能时，必须首先通过调用 create_sandbox_session() 函数创建一个新的沙箱会话。 然后，你可以使用 run_python_code() 或 run_on_command_line() 函数在沙箱上运行。 完成后，必须通过调用 stop_sandbox_session() 函数销毁沙箱会话。 注意： - 沙箱已预装常用的数据分析包，但如果不确定某个包是否存在，或由于缺少包而导致导入错误， 可以检查是否已安装，如果没有则安装它。 ”“” |
对于 tool，我们暴露使用 sandbox_manager 实例来创建、停止和执行用户沙箱中代码或命令的方法。
注意： 在以下 tool 方法的参数中，我们将包含一个 ctx: Context。这告诉 FastMCP 自动将请求上下文注入到该参数中，这个过程称为依赖注入。然后，我们通过将此上下文传递给我们的辅助方法（get_user_api_key）来获取 API 密钥（如果存在）。
| … # 以下现有代码 @mcp.tool() def create_sandbox_session(ctx: Context) -> str: “”“ 这将创建一个沙箱实例，并返回带有沙箱 ID 的成功消息或错误消息。 ”“” try: return sandbox_manager.create_sandbox_session(get_user_api_key(ctx)) except Exception as e: return e @mcp.tool() def stop_sandbox_session(sandbox_id: str, ctx: Context) -> str: “”“ 如果存在，这将销毁一个沙箱实例。 ”“” try: return sandbox_manager.stop_sandbox_session(get_user_api_key(ctx), sandbox_id) except Exception as e: return e @mcp.tool() def run_python_code(python_code: str, sandbox_id: str, ctx: Context) -> dict: “”“ 在沙箱中运行 Python 代码，如果有图像输出则跳过。 参数： python_code (str): 要运行的 Python 代码。 sandbox_id (str): 沙箱的 ID。 注意： ctx (Context) 是一个依赖注入对象，会自动传递。 返回： dict: 包含 stdout、logs、error 等。 ”“” console.print( Panel( python_code, title=“代理正在执行 Python 代码”, border_style=“blue”, ) ) try: result = sandbox_manager.run_python_code( python_code, get_user_api_key(ctx), sandbox_id ) # 显示结果。注意：仅用于测试，不要在生产中使用。 display_sandbox_code_output(result) return result except Exception as e: return {“error”: str(e)} @mcp.tool() def run_on_command_line(command_line: str, sandbox_id: str, ctx: Context) -> dict: “”“ 在沙箱中运行命令。 参数： command_line (str): 要运行的命令。 sandbox_id (str): 沙箱的 ID。 注意： ctx (Context) 是一个依赖注入对象，会自动传递。 返回： dict: 包含命令输出和执行错误（如有）。 ”“” console.print( Panel( command_line, title=“代理正在执行命令行”, border_style=“blue”, ) ) try: result = sandbox_manager.run_on_command_line( command_line, get_user_api_key(ctx), sandbox_id ) # 显示结果。注意：仅用于测试，不要在生产中使用。 display_sandbox_command_output(result) return result except Exception as e: return {“execution error”: str(e)} |
现在，我们将为其中一个演示情景暴露一个单一资源，稍后我们会看到。这个资源返回用户股票投资组合的模拟数据。与 prompt 和 tool 不同，resource 需要指定一个 URL 以便访问。
| … # 以下现有代码 @mcp.resource(“data://user_stock_portfolio”) def get_user_portfolio() -> dict: “”“ 返回用户持有的主要指数 ETF 和个股的投资组合。 返回： dict: 包含股票代码、数量和平均买入价格的投资组合 ”“” portfolio = { “holdings”: [ # 主要指数 ETF { “ticker”: “SPY”, “name”: “SPDR S&P 500 ETF”, “quantity”: 4, “avg_purchase_price”: 670.13, “asset_type”: “ETF”, }, … # 为了简洁而跳过 ] } return portfolio |
最后，我们添加启动服务器的代码。我们不会使用 stdio 传输，因为 Rich 控制台会打印到终端，这是一个阻塞操作，会干扰。相反，我们将使用 streamable-http 传输（这也是你在生产环境中会使用的，因为它最适合客户端和服务器之间通过 HTTP 进行远程连接）。
| … # 以下现有代码 if __name__ == “__main__”: # 运行服务器 # 注意：我们使用 streamable-http 作为传输协议而不是 stdio，因为我们要打印到控制台，这会阻塞 stdio。 # 在生产中，你应该使用 SSE 或 streamable-http 而不是 stdio。 asyncio.run(mcp.run(transport=“streamable-http”)) |
呼，让我们启动 MCP 服务器，这样我们就可以得到它运行的 URL，我们需要用它来连接我们的 AI 代理。在终端中运行以下命令。
| uv run mcp_server.py |
你应该看到服务器正在运行。记下它运行的 URL。
将 AI 代理与 MCP 服务器集成
现在，我们将开始在 mcp_client.py 中构建 AI 代理。mcp-use 库使这一过程变得简单。首先，我们将调试级别设置为 INFO，以便查看代理正在做什么。在 main 方法中，我们为 mcp 客户端创建一个配置字典。它指定了可用的 MCP 服务器，使用一个名称（我使用了 “stock&sandbox”）和 MCP 服务器的 URL（记得添加 /mcp），同时我们还将用户的 API 密钥作为 “auth” 的值包含进去——mcp-use 会自动将其插入到每个请求的 Authorization bearer 头中。
由于 mcp-use 依赖于 langchain-openai，我们传递 Novita 的基础 URL、API 密钥和 LLM 模型名称，因为 Novita 与 OpenAI 兼容。
除了代理的响应，我们还希望包含它在服务器上使用的沙箱 ID（如果有的话）（稍后解释原因）。为此，我们将定义一个 Pydantic 类来表示我们的响应格式。
| import asyncio import os from datetime import datetime from typing import Optional import mcp_use from dotenv import load_dotenv from langchain_openai import ChatOpenAI from mcp_use import MCPAgent, MCPClient from pydantic import BaseModel, Field from rich.console import Console from rich.panel import Panel from rich.prompt import Prompt # 加载环境变量 load_dotenv() console = Console() # 注意：1 为 INFO 级别，2 为详细 DEBUG 级别，0 为无调试输出。 mcp_use.set_debug(1) class ResponseFormat(BaseModel): response: str id_of_used_sandbox: Optional[str] = Field( …, description=“如果使用了沙箱，则为沙箱的 ID” ) async def main(model: str, base_url: str, api_key: str): # 创建配置字典 config = { “mcpServers”: { “stock&sandbox”: { # 如果 MCP 服务器运行的 URL 不同，请修改下面的内容， # 同时记得添加 /mcp。 “url”: “http://127.0.0.1:8000/mcp”, “auth”: api_key, } } } # 从配置字典创建 MCPClient client = MCPClient(config) # 创建 LLM llm = ChatOpenAI(model=model, base_url=base_url, api_key=api_key) … |
然后，我们通过传入 LLM、MCP 客户端、最大步数（限制代理在响应前可以采取的操作数量）并启用内存（以便 mcp-use 处理我们的对话历史）来创建 MCP 代理。我们还提供了一个系统提示，包含当前日期和时间，以及使用沙箱的自定义指令（我添加了这个，因为有些模型会忘记读取服务器上暴露的指令提示）。
最后，我们设置标准的对话循环：获取用户输入，传递给 agent.run()，并打印响应。如果使用了沙箱，我们将创建一个会话来手动调用服务器上的停止方法，作为一种安全措施，以防模型忘记关闭它。
| … async def main(model: str, base_url: str, api_key: str): … # 以下现有代码 # 使用客户端创建代理 agent = MCPAgent( llm=llm, client=client, max_steps=25, memory_enabled=True, # mcp-use 会自动处理对话历史。 system_prompt=f"“” 你是一个乐于助人的助手，当前日期是 {datetime.now().strftime(‘%Y-%m-%d’)} 必须记住： - 在调用任何工具之前，首先调用 instructions_for_sandbox_use() 以阅读它们。 - 确保在使用沙箱后、回复用户之前调用 stop_sandbox_session()。 “”“, ) console.print( Panel( ”[bold green]MCP 会话已开始[/bold green]\ 输入 ‘quit()’ 退出。“, title=“MCP Session”, border_style=“green”, ) ) while True: user_input = Prompt.ask(”\ [bold yellow]>>> 用户消息[/bold yellow]“) if user_input.lower().strip() == “quit()”: break # 将查询传递给代理并等待响应。 response_obj = await agent.run(user_input, output_schema=ResponseFormat) console.print( f”\ [bold green]>>> 助手回复: {response_obj.response} [/]" ) if response_obj.id_of_used_sandbox: # 如果沙箱仍然活动，将触发 MCP 服务器上的关闭操作。 session = await client.create_session(“stock&sandbox”) await session.call_tool( name=“stop_sandbox_session”, arguments={“sandbox_id”: response_obj.id_of_used_sandbox}, ) await session.disconnect() |
最后，添加启动客户端的代码：
| … # 以下现有代码 if __name__ == “__main__”: asyncio.run( main( model=“qwen/qwen3-coder-480b-a35b-instruct”, base_url=os.getenv(“NOVITA_BASE_URL”), api_key=os.getenv(“NOVITA_API_KEY”), ) ) |
测试运行我们的 MCP AI 代理：
你可以通过在终端中运行以下命令来启动 AI 代理应用程序：
| uv run mcp_client.py |
以下是对以下提示进行演示运行的视频链接：
- 用户提示 1： “我有 2000 美元。从 yfinance 获取过去 6 个月美国主要股指的表现，并运行机器学习模型来预测如何配置这笔投资，以在未来 2 个月内最大化潜在回报。”（在此处插入链接）
- 用户提示 2： “运行多次美国经济通缩崩溃的模拟，选择最可能的一种，并解释它将如何影响我的股票投资组合。”（在此处插入链接）
生产环境中 MCP 服务器的提示
虽然本教程重点介绍了如何使用 Novita Sandbox 构建一个可工作的 MCP 服务器，但部署到生产环境需要额外考虑：
使用正确的传输方式： 虽然 “stdio” 适用于本地开发，但生产环境的 MCP 服务器应使用 “streamable-http” 以启用远程连接，就像我们上面所做的那样。
实现身份验证： 像我们上面所做的那样，确保通过身份验证保护你的 MCP 服务器端点。确保每个客户端只能访问他们需要的工具和资源。你可以在 FastMCP authentication 阅读更多身份验证方法。
启用日志记录： 使用日志记录器监控服务器活动、调试问题并跟踪使用模式。这对维护和故障排除至关重要。
速率限制和配额： 通过实施速率限制和配额来保护你的服务器免受滥用。这在使用资源密集型工具时尤其重要。
文档和版本控制： 保持 MCP 服务器 API 和版本控制的清晰文档，以方便开发人员和 LLM 集成。
结论
呼，你现在终于可以构建一个 MCP 服务器，AI 代理可以通过自然语言指令远程执行代码——这是 Software 3.0 的实际实现。
在本教程中，你学习了如何构建一个具有代码执行能力的 MCP 服务器，管理沙箱生命周期，并使用 mcp-use 创建一个连接到你的服务器的 AI 代理。接下来，尝试扩展它，添加数据库访问、网页搜索，或将多个服务器链接到一个代理上。前往 Novita，我们拥有构建 AI 代理所需的工具。
Novita AI 是一个 AI 云平台，为开发者提供通过简单 API 部署 AI 模型的简便方式，同时提供经济实惠且可靠的 GPU 云服务，用于构建和扩展您的 AI 应用。
