MCP Run Python
MCP Run Python 包是一个 MCP 服务器,允许代理在安全、沙箱环境中执行 Python 代码。它使用 Pyodide 在 JavaScript 环境中运行 Python 代码,从而将执行与主机系统隔离。
特性
- 安全执行:在沙箱 WebAssembly 环境中运行 Python 代码
- 包管理:自动检测和安装所需的依赖项
- 完整结果:捕获标准输出、标准错误和返回值
- 异步支持:正确运行异步代码
- 错误处理:提供详细的错误报告以进行调试
安装
MCP Run Python 服务器作为 NPM 包 分发,可以直接使用 npx
运行
npx @pydantic/mcp-run-python [stdio|sse]
其中
stdio
:使用 stdin/stdout 传输(用于子进程用法)运行服务器sse
:使用 HTTP Server-Sent Events 传输(用于远程连接)运行服务器
在 客户端 文档中描述了 @pydantic/mcp-run-python
与 PydanticAI 的用法。
直接使用
除了将此服务器与 PydanticAI 一起使用外,它还可以连接到其他 MCP 客户端。为了清晰起见,在此示例中,我们直接使用 Python MCP 客户端 进行连接。
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
code = """
import numpy
a = numpy.array([1, 2, 3])
print(a)
a
"""
async def main():
server_params = StdioServerParameters(
command='npx', args=['-y', '@pydantic/mcp-run-python', 'stdio']
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
print(len(tools.tools))
#> 1
print(repr(tools.tools[0].name))
#> 'run_python_code'
print(repr(tools.tools[0].inputSchema))
"""
{'type': 'object', 'properties': {'python_code': {'type': 'string', 'description': 'Python code to run'}}, 'required': ['python_code'], 'additionalProperties': False, '$schema': 'https://json-schema.fullstack.org.cn/draft-07/schema#'}
"""
result = await session.call_tool('run_python_code', {'python_code': code})
print(result.content[0].text)
"""
<status>success</status>
<dependencies>["numpy"]</dependencies>
<output>
[1 2 3]
</output>
<return_value>
[
1,
2,
3
]
</return_value>
"""
如果发生异常,status
将为 install-error
或 run-error
,并且 return_value
将被 error
替换,其中将包含回溯和异常消息。
依赖项(Dependencies)
依赖项在代码运行时安装。
依赖项可以通过两种方式定义
从导入推断
如果没有元数据,则依赖项从代码中的导入推断,如上面的示例所示。
内联脚本元数据
正如 PEP 723 中介绍的,此处解释,并由 uv 推广——依赖项可以在文件顶部的注释中定义。
这允许使用代码中未导入的依赖项,并且更加明确。
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
code = """\
# /// script
# dependencies = ["pydantic", "email-validator"]
# ///
import pydantic
class Model(pydantic.BaseModel):
email: pydantic.EmailStr
print(Model(email='hello@pydantic.dev'))
"""
async def main():
server_params = StdioServerParameters(
command='npx', args=['-y', '@pydantic/mcp-run-python', 'stdio']
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool('run_python_code', {'python_code': code})
print(result.content[0].text)
"""
<status>success</status>
<dependencies>["pydantic","email-validator"]</dependencies>
<output>
email='hello@pydantic.dev'
</output>
"""
它还允许为非二进制包锁定版本(Pyodide 仅支持其支持的二进制包的单个版本,如 pydantic
和 numpy
)。
例如,您可以将依赖项设置为
# /// script
# dependencies = ["rich<13"]
# ///
日志记录
MCP Run Python 支持将 python 执行中的 stdout 和 stderr 作为 MCP 日志消息 发出。
为了发出日志,您必须在连接到服务器时设置日志记录级别。默认情况下,日志级别设置为最高级别 emergency
。
目前,由于 Python MCP 客户端中的一个错误,无法演示这一点,请参阅 modelcontextprotocol/python-sdk#201。