使用说明

python_executor_mcp（python脚本执行器）

MCP（模型上下文协议）服务器，用于执行Python脚本和内联代码，具有：

紫外线管理的瞬态环境
显式依赖解析
环境变量支持（dict或.env文件）
标记第一MCP工具响应(markdown_output=True 默认情况下；集 markdown_output=False 对于结构化有效载荷）
智能异步模式 -自动超时切换到后台
同步或异步（后台）执行
stdout/stderr的可选流式传输
带有进度更新的作业跟踪
基准测试（墙时间、CPU时间、峰值内存）
作业生命周期的反思与控制

此自述文件记录了服务器的工具、资源模式、数据模型、使用示例和操作指南。

1.功能概览

能力	同步	异步	智能异步	流式传输	依赖关系	基准测试	环境变量	进度
运行目录	中的现有脚本✅	✅ （异步变体）	❌	✅ （当流=True时）	❌	❌	✅	❌
在目录	中运行内联脚本✅	✅	❌	✅	❌	❌	✅	❌
运行带有依赖项的代码	✅	✅	❌	✅	✅	✅ （基准脚本）	✅	❌
基准瞬态执行	✅	❌	❌	❌	✅	✅	✅	❌
智能异步作业管理	N/A	N/A	✅	N/A	N/A	N/A	N/A	N/A
列出具有筛选功能的作业	✅	N/A	N/A	N/A	N/A	N/A	N/A	N/A
获取作业状态和进度	✅	N/A	适用于智能异步	N/A	N/A	N/m	N/A	N/A
取消正在运行的作业	✅	N/A	适用于智能异步	N/A	N/A	N/m	N/A	N/A
精简旧工作	✅	N/A	N/A	N/A	N/A	N/A	N/A	N/A

2.要求

Python>=3.13
uv（用于环境分辨率）
fastmcp（服务器框架）
psutil（基准指标）
pydantic（响应模型）
python dotenv（环境变量加载）

安装和同步环境：

uv sync

3.运行服务器

明确选择StdIO传输（MCP控制台集成的默认设置）：

uv run python-executor-mcp

或者使用模块：

uv run python -m python_mcp_server

兼容性别名：

uv run python-mcp-server

如果从编辑器（例如Zed）调用并需要一个固定的工作目录，请创建一个启动器脚本：

#!/usr/bin/env sh
cd /home/torstein.sornes/code/python-mcp
exec uv run python-executor-mcp

4.工具参考

4.1同步执行

run_script_in_dir（标签：执行、同步）

使用uv或系统Python在目标目录中执行Python脚本（现有文件或内联内容）。

参数：

目录：路径（必须存在）
script_path：路径|无（绝对或相对于目录；与script_content互斥）
script_content:str|None（内联源代码；与script_path互斥；自动保存在此仓库下 autosaved_scripts/ 使用JSON参数sidecar）
args：列表〔str〕|无
use_uv:bool（真使用 uv run)
python_version:str|无（完全次要；仅当use_uv=True时才生效）
timeout_seconds：int（0=无限制）
env_vars:dict[str，str]|无（要设置的环境变量）
env_file:路径|无（要加载的.env文件的路径）

返回（RunScriptResult）：

{
  "stdout": str,
  "stderr": str,
  "exit_code": int,
  "execution_strategy": "uv-run" | "system-python",
  "elapsed_seconds": float
}

run_script_with_dependencies（标签：执行、依赖、同步）

在具有显式依赖关系的瞬态uv环境中执行内联代码或现有脚本。

参数：

script_content或script_path（互斥；内联 script_content 在此仓库下自动保存 autosaved_scripts/ 使用JSON参数sidecar）
python_version:str（完全次要）
依赖关系：list[str]|无（PEP 440说明符）
args：列表〔str〕|无
timeout_seconds：int（0=无限制）
env_vars:dict[str，str]|无（要设置的环境变量）
env_file:路径|无（要加载的.env文件的路径）

返回（RunWithDepsResult）：

{
  "stdout": str,
  "stderr": str,
  "exit_code": int,
  "execution_strategy": "uv-run",
  "elapsed_seconds": float,
  "resolved_dependencies": list[str],
  "python_version_used": str
}

保存脚本自检

py_list_scripts(include_autosaved=False)：列出此仓库中手动保存的脚本 scripts/；set include_autosaved=True 还显示自动保存的内联脚本。
py_view_script(script_name, autosaved=False)：默认情况下查看手动脚本；集 autosaved=True 查看自动保存的内联脚本及其元数据sidecar。

benchmark_script（标签：基准、性能、同步）

在收集指标的同时，执行具有依赖关系解析的代码或脚本。

参数：

script_content或script_path（内联 script_content 在此仓库下自动保存 autosaved_scripts/ 使用JSON参数sidecar）
python版本：str
依赖关系：list[str]|无
args：列表〔str〕|无
超时秒数：int
sample_interval:float（内存轮询间隔，单位为秒；默认值为0.05）
env_vars:dict[str，str]|无（要设置的环境变量）
env_file:路径|无（要加载的.env文件的路径）

回报（基准结果）：

{
  "stdout": str,
  "stderr": str,
  "exit_code": int,
  "execution_strategy": "uv-run",
  "elapsed_seconds": float,
  "resolved_dependencies": list[str],
  "python_version_used": str,
  "wall_time_seconds": float,
  "cpu_time_seconds": float,
  "peak_rss_mb": float
}

4.2异步执行

异步变量将关注点分开，并立即返回作业描述符。

run_script_in_dir_async（标签：执行、异步、流）

与run_script_in_dir类似的参数，除了：

省略timeout_seconds（没有内置超时；如果需要，建议使用外部kill_job）
stream:bool（启用定期的stdout/stderr收获）
env_vars:dict[str，str]|无（要设置的环境变量）
env_file:路径|无（要加载的.env文件的路径）

返回（AsyncJobStart）：

{
  "job_id": str,
  "status": "started",
  "execution_strategy": "uv-run" | "system-python"
}

run_script_with_dependences_async（标签：执行、依赖、异步、流）

类似于run_script_with_dependences的参数加上：

stream:bool（启用定期的stdout/stderr收获）
env_vars:dict[str，str]|无（要设置的环境变量）
env_file:路径|无（要加载的.env文件的路径）

返回（AsyncDepsJobStart）：

{
  "job_id": str,
  "status": "started",
  "python_version_used": str,
  "resolved_dependencies": list[str]
}

4.3工作管理与反思

list_running_jobs（标签：作业、自省）

返回作业字典列表：

[
  {
    "job_id": str,
    "running": "True" | "False",
    "exit_code": str | "None",
    "pid": str,
    "elapsed_seconds": str,
    "stream": "True" | "False",
    "stdout_chunks": str,
    "stderr_chunks": str
  },
  ...
]

get_job_output（标签：作业、自省）

检索当前或最终的作业输出。如果作业已完成但尚未完成，则在此处完成。

返回有效载荷：

{
  "status": "running" | "finished",
  "stdout": str,
  "stderr": str,
  "exit_code": str | "None",
  "elapsed_seconds": str
}

kill_job（标签：作业，控件）

终止正在运行的进程；完成输出。

退货：

{
  "job_id": str,
  "status": "killed",
  "exit_code": str
}

4.4流媒体资源

作业流：//{job_id}

快照资源提供增量组合输出。格式：

---STDOUT---
<current stdout>
---STDERR---
<current stderr>

未完成作业。
对于增量处理，在客户端跟踪之前的长度。

5.数据模型

Pydantic模型（简化）：

运行脚本结果
异步作业启动
RunWithDeps结果
Asyncdepsjobstart
基准测试结果

这些模型允许客户端推断精确的JSON模式（名称、类型、可选性）并减少不匹配（例如，整数与字符串）。

6.智能异步模式

服务器实现 智能异步模式 来自mcpbuilder的智能后台作业执行技能，具有自动超时处理功能。

6.1工作原理

自动超时切换：

装饰有 @smart_async 尝试在超时预算内同步完成
如果操作超过超时，它会自动切换到后台执行
基本任务是 屏蔽的 为了防止取消，它继续无缝运行
快速操作，在线完成，无开销
慢速操作会立即返回job_id进行跟踪

显式异步模式：

使用 async_mode=True 用于在后台立即启动作业的参数
适用于已知的长时间运行操作
跳过超时尝试并立即返回job_id

进度跟踪：

后台作业可以报告进度更新
进度存储在作业元数据中并持久化到磁盘
使用 create_progress_callback() 在智能异步工具中

6.2作业管理工具

获取作业状态：

py_job_status(job_id="abc-123")
# Returns: {"job": {"status": "running", "progress": {"current": 5, "total": 10}}}

列出作业：

py_list_jobs(status_filter="running", limit=50)
# Returns: {"jobs": [...], "total": 5}

取消作业：

py_cancel_job(job_id="abc-123")
# Returns: {"job_id": "abc-123", "status": "cancelled"}

修剪旧工作：

py_prune_jobs(keep_completed=False, max_age_hours=24)
# Returns: {"removed": 10, "remaining": 5}

6.3作业生命周期

待定 -作业已创建但未启动
跑步 -作业执行
完成 -作业已成功完成
失败 -作业引发异常
已取消 -作业已被用户取消

6.4工作坚持性

作业保存到 ~/.python_mcp/meta/jobs.json
服务器重启后仍能存活
在重新启动期间运行的作业在下次启动时标记为“失败”
进度更新自动保持不变

6.5示例：创建智能异步工具

from python_mcp_server.smart_async import smart_async, create_progress_callback

@smart_async(timeout_env="MY_TIMEOUT", default_timeout=30.0)
async def my_long_task(
    items: list[str],
    async_mode: bool = False,
    job_label: str | None = None
) -> dict:
    """Process items with automatic background switching."""
    progress = create_progress_callback()
    
    results = []
    for i, item in enumerate(items):
        result = await process_item(item)
        results.append(result)
        progress(i + 1, len(items), f"Processed {i + 1} items")
    
    return {"results": results, "total": len(results)}

用途：

# Fast - completes synchronously
result = await my_long_task(items=["a", "b", "c"])
# Returns: {"results": [...], "total": 3}

# Slow - switches to background automatically
result = await my_long_task(items=long_list)
# Returns: {"job_id": "...", "status": "running", "message": "..."}

# Explicit async - launches immediately
result = await my_long_task(items=items, async_mode=True)
# Returns: {"job_id": "...", "status": "pending"}

7.环境变量

所有执行工具都支持通过两种方法设置环境变量：

6.1使用env_vars字典

直接将环境变量作为字典传递：

run_script_in_dir(
  directory="/work/project",
  script_path="main.py",
  env_vars={
    "DATABASE_URL": "postgresql://localhost/db",
    "API_KEY": "secret123",
    "DEBUG": "true"
  }
)

6.2使用.env文件

从.env文件加载环境变量：

run_script_with_dependencies(
  script_content="import os; print(os.getenv('MY_VAR'))",
  python_version="3.13",
  env_file="/path/to/.env"
)

.env文件格式：

使用标准 python dotenv 格式：

# Comments are supported
DATABASE_URL=postgresql://localhost/testdb
API_KEY="secret123"
DEBUG=true
QUOTED='single quotes work too'
MULTILINE_VAR="line1
line2
line3"

支持：

评论(#)
单引号和双引号
多行值
变量扩展
空行

6.3优先权

当两者都 env_file 和 env_vars 如果提供了，它们将与此优先级合并（稍后覆盖较早）：

当前流程环境（继承自MCP服务器）
变量来自 env_file
变量来自 env_vars dict（最高优先级）

例子：

# .env contains: MY_VAR=from_file
run_script_in_dir(
  directory="/work",
  script_path="test.py",
  env_file="/work/.env",
  env_vars={"MY_VAR": "from_dict"}  # This overrides the .env value
)
# Result: MY_VAR=from_dict

8.使用示例

8.1同步运行（现有脚本）

run_script_in_dir(
  directory="/work/project",
  script_path="main.py",
  use_uv=True,
  python_version="3.13",
  timeout_seconds=300
)

8.2同步内联代码

run_script_in_dir(
  directory="/tmp",
  script_content="print('Hello world')",
  use_uv=False
)

8.3依赖关系（同步）

run_script_with_dependencies(
  script_content="import requests; print(requests.__version__)",
  python_version="3.13",
  dependencies=["requests"]
)

8.4基准

benchmark_script(
  script_content="print(sum(i*i for i in range(500000)))",
  python_version="3.13",
  dependencies=[]
)

8.5异步+流媒体

start = run_script_in_dir_async(
  directory="/opt/jobs",
  script_path="long_task.py",
  use_uv=True,
  python_version="3.13",
  stream=True
)

job_id = start.job_id

# Poll for partial output
snapshot = get_resource("job-stream://{job_id}")

# Final consolidated output
final = get_job_output(job_id)

8.6与依赖项异步

async_start = run_script_with_dependencies_async(
  script_content="import time; [print(i) or time.sleep(0.2) for i in range(5)]",
  python_version="3.13",
  dependencies=[],
  stream=True
)

8.7环境变量

run_script_with_dependencies(
  script_content="""
import os
print(f"DB: {os.getenv('DATABASE_URL')}")
print(f"Key: {os.getenv('API_KEY')}")
""",
  python_version="3.13",
  dependencies=[],
  env_vars={
    "DATABASE_URL": "postgresql://localhost/mydb",
    "API_KEY": "sk-test-123"
  }
)

8.8智能异步作业跟踪

# Tool returns job_id when it switches to background
result = await some_long_operation()

if "job_id" in result:
    # Poll for status
    status = py_job_status(job_id=result["job_id"])
    print(status["job"]["status"])  # "running"
    print(status["job"]["progress"])  # {"current": 50, "total": 100}
    
    # Wait for completion
    while status["job"]["status"] == "running":
        await asyncio.sleep(1)
        status = py_job_status(job_id=result["job_id"])
    
    # Get final result
    print(status["job"]["result"])

8.9丢掉工作

kill_job(job_id="abc123")

9.错误处理

常见错误：

错误	原因	恢复
FileNotFoundError	目录、脚本路径或env_file缺失	检查路径解析/权限
ValueError	提供了script_path和script_content两者或两者都没有	只提供一个
超时（同步工具）	执行超过Timeout_seconds	使用更大的超时重试或优化脚本
OSError/子进程错误	解释器或uv不可用	确保uv已安装并位于PATH中
psutil。错误（基准）	采样前流程已结束或访问被拒绝	可以忽略；部分指标可用

超时行为：

进程已终止。
stderr后缀为 [TIMEOUT].
exit_code反映了操作系统的终止代码。

10.流媒体语义

stream=True通过内部后台任务激活定期轮询。
作业流资源在每次轮询时返回完整的累积输出。
对于增量处理，从外部跟踪之前的长度。
进程完成后，快照稳定。

未来的增强功能（计划中）：

基于游标的流式传输（仅返回新字节）。
用于推送流的SSE/HTTP传输变体。
日志截断阈值，以避免巨大的有效载荷。

11.性能说明

由于分辨率的原因，具有许多依赖性的冷紫外运行可能会增加延迟；考虑缓存未来的临时环境。
STREAM_POLLINTERVAL当前为0.2秒（响应性和CPU开销之间的平衡）。
psutil采样间隔可配置（benchmark_script中的sample_interval）。
内联 script_content 保存在此存储库中 autosaved_scripts/ 目录与a .json 包含MCP工具参数的sidecar；git会忽略该目录。

12.安全考虑

根据项目指令：无沙盒。执行的代码有：

服务器用户下的完全文件系统访问
网络接入
生成子流程的能力

如果您以后需要限制：

添加可选标志以禁用网络（iptables/防火墙/包装器）
使用资源限制（ulimit或cgroups）
添加代码长度和参数验证

环境变量：

环境变量直接传递给子流程
.env文件在客户端解析（对于共享文件系统中的机密不安全）
考虑对敏感凭据使用秘密管理工具

13.可扩展性路线图

未来的潜在工具：

format_code（黑色/菱形）
统计分析（流氓+强盗）
environment_cache_list/environment_create_persistent
job_cleanup（删除已完成的作业和临时文件）
stream_deltas（基于光标）
run_script_profile（cProfile统计数据）

14.故障排除

症状	原因	解决方案
启动服务器时编辑器超时	混合路径/不正确的工作目录	确保服务器在项目根目录中启动时没有冲突--目录标志
验证错误（类型不匹配）	使用旧模式/以前的dict[str，str]设计	使用更新的Pydantic模式感知客户端
流式传输中未捕获输出	异步启动时stream=False	设置stream=True或使用sync变量
基准返回零peak_rss_mb	进程时间太短，无法采样	增加工作负载或减少sample_interval

|环境变量在脚本中不可见|键名中的拼写错误或.env未加载|检查env_vars字典键是否匹配用法；验证env文件路径|

15.设计原则概述

script_path/script_content的互斥性减少了歧义。
同步和异步工具分开，以实现更清晰的模式契约。
结构化Pydantic模型可以防止隐式类型强制错误。
以资源为中心的流式传输（作业流）使数据检索变得简单。
日志记录生命周期事件（启动、异步启动、终止、完成）。
智能异步模式可实现最佳性能（内联快速任务，后台慢速任务）

16.组合工作流示例

# 1. Fire off a dependency-heavy async script
start = run_script_with_dependencies_async(
  script_content="import numpy as np; import time; [print(np.sqrt(i)) or time.sleep(0.3) for i in range(10)]",
  python_version="3.13",
  dependencies=["numpy"],
  stream=True
)

# 2. Periodically poll streaming output
for _ in range(5):
    chunk = get_resource(f"job-stream://{start.job_id}")
    print(chunk)
    status = get_job_output(start.job_id)
    if status["status"] == "finished":
        break
    time.sleep(1.0)

# 3. Final result
final = get_job_output(start.job_id)
print(final["stdout"])

17.许可/归属

请参阅项目级许可证（如果添加）。FastMCP框架文档：https://gofastmcp.com

18.维护检查表

在合并更改之前：

验证文档字符串是否与异步/同步分段匹配。
确认Pydantic模型反映了实际的运行时负载。
测试超时边缘情况（timeout_seconds=1）。
使用小型和大型工作负载练习benchmark_script。

19.常见问题

Q：为什么要将异步工具与同步工具分开？ A：返回模式根本不同（作业描述符与最终输出）。分离避免了输出过载和模式模糊。

Q：我可以流式传输同步运行吗？ A：否--同步等待完成。使用异步变量+stream=True。

Q：如何截断巨大的stdout？ A：目前尚未内置；实现包装器或添加增强功能以限制存储的块大小。

Q：为什么只有Python的次要版本？ A：简化解析逻辑，避免歧义的解释器选择；如果缺失，uv将负责安装。

Q：我可以同时使用env_vars和env_file吗？ A：是的，它们与优先于env_file值的env_vars合并。

Q：环境变量对生成的子流程可见吗？ A：是的，它们被传递给子流程环境，并由任何子流程继承。

Q：什么是智能异步模式？ A：一种基于装饰器的方法，其中工具尝试在超时内同步完成，然后在超时后自动切换到后台执行。任务被屏蔽以防止取消。

Q：如何在我的工具中使用智能异步？ A：用以下命令装饰你的异步函数 @smart_async()，添加 async_mode 和 job_label 参数和使用 create_progress_callback() 报告进展情况。

Q：我可以跟踪后台作业的进度吗？ A：是-使用 create_progress_callback() 在您的工具和调用中 py_job_status() 检查进度。

20.贡献

拟议的延期指南：

为任何新的输出类型添加新的Pydantic模型。
提供完整的文档字符串，包括：摘要、参数、返回、错误、示例。
适当标记工具以帮助发现。
保持一致的命名（script_path/script_content）。

快乐的建筑。使用结构化工具在MCP环境中实现可靠、可观察的脚本执行工作流。