传统的MCP工具有个根本限制:只能运行一次性命令。你让它执行python manage.py runserver
,它会立即退出,因为无法管理长时间运行的进程。这意味着AI永远看不到服务器的实际运行状态,只能生成代码然后盲目地希望它能工作。
核心突破:输出重定向
解决这个问题的关键在于将服务器的所有输出都重定向到文件中,让AI能够随时查看。不仅是stdout,stderr也必须重定向,因为大多数关键信息(错误、警告、调试信息)都在stderr中。
async def start_long_process(self, project_path: str, command: str, process_name: str, output_file: str = None):
"""启动长时间运行的进程,将所有输出重定向到文件"""
# 准备输出重定向文件
output_file_handle = None
if output_file:
project_dir = Path(project_path)
output_path = project_dir / output_file
output_file_handle = open(output_path, 'w', encoding='utf-8')
# 启动进程 - 关键点:stderr也重定向到同一个文件
process = await asyncio.create_subprocess_shell(
wrapped_command,
cwd=project_path,
stdout=output_file_handle or asyncio.subprocess.PIPE,
stderr=output_file_handle, # 这里是重点:错误输出也进入文件
env=env
)
# 保存进程信息
process_info = {
'process': process,
'start_time': datetime.now(),
'output_file_handle': output_file_handle, # 保持文件句柄开放
'project_path': project_path
}
self.active_processes[process_name] = process_info
这样做的效果是,Django服务器的所有输出都会实时写入文件:
System check identified no issues (0 silenced).
December 13, 2024 - 15:32:41
Django version 4.2.7, using settings 'library_system.settings'
Starting development server at http://127.0.0.1:8000/
Quit the server with CTRL-BREAK.
[13/Dec/2024 15:32:45] "GET /api/books/ HTTP/1.1" 200 2
[13/Dec/2024 15:32:46] "POST /api/books/ HTTP/1.1" 201 89
[13/Dec/2024 15:32:47] "GET /admin/ HTTP/1.1" 200 19737
实时输出查看
AI可以随时查看服务器状态:
def read_process_output(self, project_path: str, output_file: str, last_lines: int = 10):
"""读取进程输出文件的最新内容"""
project_dir = Path(project_path)
file_path = project_dir / output_file
content = file_path.read_text(encoding='utf-8')
if last_lines > 0:
lines = content.split('\n')
content = '\n'.join(lines[-last_lines:])
return {
'success': True,
'content': content,
'file_size': file_path.stat().st_size
}
优雅的进程终止
终止进程时需要正确关闭文件句柄,避免资源泄漏:
def terminate_process(self, process_name: str):
"""终止进程并清理资源"""
process_info = self.active_processes[process_name]
process = process_info['process']
# 先尝试优雅关闭
if process.returncode is None:
process.terminate()
# 等待5秒,如果还没退出就强制杀死
try:
loop = asyncio.get_event_loop()
if not loop.is_running():
loop.run_until_complete(
asyncio.wait_for(process.wait(), timeout=5.0)
)
except asyncio.TimeoutError:
process.kill()
# 关闭文件句柄
if process_info.get('output_file_handle'):
process_info['output_file_handle'].close()
# 从管理列表中移除
del self.active_processes[process_name]
实际效果
在Django项目开发中,这个机制的效果非常明显:
- 启动服务器:
start_server_process("python manage.py runserver 8000", "django_server", "server.log")
- 实时监控:AI可以通过
read_process_output("server.log", 10)
看到最新的请求日志 - 发现问题:如果有错误,AI立即能从日志中看到详细的错误信息
- 修复验证:修改代码后,Django自动重载,AI能看到重载成功的日志
- 功能测试:AI执行API测试,能从日志中看到实际的HTTP请求和响应
虚拟环境集成
所有进程都自动在项目虚拟环境中运行:
def _wrap_command_for_venv(self, command: str, project_path: str) -> str:
"""将命令包装为在虚拟环境中执行"""
paths = self._get_venv_paths(project_path)
if command.startswith("python "):
if paths['python_exe'].exists():
return command.replace("python ", f'"{paths["python_exe"]}" ', 1)
return command
编码问题解决
通过环境变量和正确的解码方式解决跨平台编码问题:
def _prepare_environment(self, project_path: str) -> Dict[str, str]:
"""准备执行环境"""
env = os.environ.copy()
env['PYTHONIOENCODING'] = 'utf-8'
env['PYTHONUNBUFFERED'] = '1' # 确保输出不被缓冲
return env
# 解码输出时使用容错机制
stdout_text = stdout.decode('utf-8', errors='replace')
实际开发体验
在Django图书管理系统的开发过程中,这个机制让AI能够:
- 看到服务器启动成功的确认信息
- 监控API请求的实时日志
- 发现模型迁移的问题并修复
- 验证前端页面的访问情况
- 观察到代码修改后的自动重载
评论区