分享一个python做的MCP应用

前言

在deepin论坛发现UOS AI最近集成了MCP,然后还有个MCP场景大发现活动，地址,就试了试，万一赢了个小鼠标呢。
所以就找llm帮忙做了一个mcp应用，调用本地ollama的。

介绍

功能很简单，输入需求，如

请帮我 xxx

然后该mcp应用会实现以下功能

• 调用本地ollama模型来根据xxx生成相应python代码
• 创建虚拟环境或使用已创建的虚拟环境
• 自动安装依赖或你的llm通过输入 pip install xxx 来帮忙安装依赖
• 执行代码，并发出日志

这样就实现了一个几乎可以帮你完成任何任务的mcp工具！

运行时会创建一个文件夹放生成的py,同时会使用一个共用的env，deepseek或其他llm在执行命令发现依赖问题时会尝试pip install xxx，把这个指令单独弄了出来识别并执行

---

简单演示

图中为Deepseek+UOS AI+这个mcp服务

svg图片换色

把一个蓝色的b站小图标换为粉色

文字生成

在一个文本里生成一些废话

网络功能

抓取bing美图到本地

复杂任务

将MiSans字体中的天影大侠提取出来绘制为白色图片并添加黑色边框

当然python有无限的可能

---

代码

import os
import re
import sys
import json
import logging
import subprocess
import datetime
from pathlib import Path
from typing import Dict, Any, Optional, Set, Tuple

# 导入 MCP 核心库
from mcp.server.fastmcp import FastMCP, Context

# 导入 LangChain 和 Pydantic
from langchain_community.llms import Ollama
from pydantic import BaseModel, Field

# --- 日志配置 ---
logger = logging.getLogger(__name__)

# --- MCP 服务器定义 ---
mcp = FastMCP(name="SkyShadowHero Task Execution Server")

# --- Pydantic 返回模型 ---
class ExecutionStage(BaseModel):
    code_generation: str
    dependency_installation: str
    execution: str

class ExecutionResult(BaseModel):
    stages: ExecutionStage
    code: str
    output: str
    error: str
    work_dir: str
    returncode: int

class ServerInfo(BaseModel):
    name: str
    model: Optional[str]
    status: str

class CommandOutput(BaseModel):
    status: str
    result: ExecutionResult
    server: ServerInfo

# --- 配置管理 ---
class MCPConfig:
    _instance = None
    config_path = Path(__file__).parent / "mcp_config.json"

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.load_config()
        return cls._instance

    def load_config(self):
        if not self.config_path.exists():
            self.config = {
                "mcpServers": {
                    "default": {
                        "model": "deepseek-coder-v2:16b",
                        "task_templates": {
                            "file_operations": {
                                "prompt": (
                                    "你是一个顶级的Python自动化专家。你的任务是根据用户的自然语言指令，生成一段完整、健壮、可直接在标准Python环境中执行的脚本。你拥有完全的创作自由，但必须严格遵守以下规范。\n\n"
                                    "## 用户指令:\n"
                                    "{task}\n\n"
                                    "## 代码生成规范 (必须严格遵守):\n"
                                    "1.  **【代码纯净性】**: 你的输出必须是纯粹的Python代码。绝对禁止包含任何Markdown标记，尤其是 ` ```python ` 和 ` ``` `。\n"
                                    "2.  **【依赖声明】**: 如果代码需要任何第三方库 (例如 `requests`, `pandas`)，必须在代码的最开始，使用 `# REQUIRE: <package_name>` 的格式进行声明。**每个依赖独立一行**。如果不需要任何第三方库，则完全不要写 `# REQUIRE:` 注释。\n"
                                    "3.  **【日志记录】**: 必须使用Python的 `logging` 模块。在脚本开始处配置好 `basicConfig`，确保日志同时输出到控制台(stdout)和当前工作目录下的 `task.log` 文件。在关键步骤和任何 `except` 块中，都必须使用 `logging.info()` 或 `logging.error()` 进行记录。\n"
                                    "4.  **【错误处理】**: 所有可能失败的操作都必须被包含在 `try...except Exception as e:` 块中。\n"
                                    "5.  **【成功信号】**: 在脚本所有操作成功完成的最后，必须调用 `print(\"任务成功完成\")`。\n"
                                    "6.  **【完整性】**: 生成的代码必须是完整的、自包含的，包含所有必要的 `import` 语句。\n\n"
                                    "现在，请根据用户指令生成代码。"
                                ),
                            }
                        }
                    }
                }
            }
            self.save_config()
        else:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                self.config = json.load(f)

    def save_config(self):
        with open(self.config_path, 'w', encoding='utf-8') as f:
            json.dump(self.config, f, indent=2, ensure_ascii=False)

    def get_server(self, name: str) -> Optional[Dict[str, Any]]:
        return self.config.get("mcpServers", {}).get(name)

# --- 【最终修复架构】核心逻辑封装 ---
class TaskWorkflow:
    def __init__(self):
        self.config = MCPConfig().config
        self.llm_cache = {}
        self.standard_libs = self._get_standard_libs()
        script_dir = Path(__file__).parent.resolve()
        self.shared_work_dir = script_dir / "mcp_tasks"
        self.shared_work_dir.mkdir(exist_ok=True)
        
        logger.info("正在初始化并检查共享虚拟环境...")
        try:
            self.venv_path = self.shared_work_dir / "venv"
            self.python_executable, self.pip_executable = self._create_virtual_env(str(self.venv_path))
            logger.info(f"共享虚拟环境已就绪。Python: {self.python_executable}, Pip: {self.pip_executable}")
        except Exception as e:
            logger.error(f"初始化共享虚拟环境失败: {e}", exc_info=True)
            raise RuntimeError(f"无法创建或验证共享虚拟环境，服务无法启动。错误: {e}")

    def _get_standard_libs(self) -> Set[str]:
        common_libs = {'os', 'sys', 'json', 're', 'logging', 'subprocess', 'pathlib', 'datetime', 'time', 'math', 'random', 'collections', 'itertools', 'functools', 'glob', 'shutil', 'tempfile', 'argparse', 'typing', '__future__'}
        if sys.version_info >= (3, 10):
            try:
                from sys import stdlib_module_names
                return set(stdlib_module_names)
            except ImportError:
                return common_libs
        return common_libs

    async def get_llm(self, model_name: str) -> Ollama:
        if model_name not in self.llm_cache:
            logger.info(f"正在加载模型: {model_name}")
            self.llm_cache[model_name] = Ollama(model=model_name, temperature=0.1, top_p=0.9, timeout=300)
        return self.llm_cache[model_name]

    def _create_virtual_env(self, venv_path_str: str) -> Tuple[str, str]:
        venv_path = Path(venv_path_str)
        if sys.platform == "win32":
            python_exe = venv_path / 'Scripts' / 'python.exe'
            pip_exe = venv_path / 'Scripts' / 'pip.exe'
        else:
            python_exe = venv_path / 'bin' / 'python'
            pip_exe = venv_path / 'bin' / 'pip'

        if not python_exe.exists() or not pip_exe.exists():
            logger.info(f"共享虚拟环境不完整或不存在，正在创建于: {venv_path_str}")
            try:
                subprocess.run(
                    [sys.executable, "-m", "venv", venv_path_str],
                    check=True, capture_output=True, text=True, timeout=120
                )
            except subprocess.CalledProcessError as e:
                logger.error(f"创建虚拟环境失败: {e.stderr}")
                raise RuntimeError(f"创建虚拟环境失败，错误: {e.stderr}")
        
        if not python_exe.exists() or not pip_exe.exists():
             raise FileNotFoundError(f"虚拟环境创建后，未找到 Python/Pip 可执行文件。")
        
        logger.info("虚拟环境验证成功。")
        return str(python_exe), str(pip_exe)

    def _post_process_code(self, generated_code: str) -> Tuple[str, Set[str]]:
        cleaned_code = re.sub(r"```python\n|```", "", generated_code).strip()
        required_deps = set(re.findall(r"#\s*REQUIRE:\s*(\S+)", cleaned_code))
        final_code = "\n".join([line for line in cleaned_code.split('\n') if not line.strip().startswith("# REQUIRE:")])
        logger.info(f"代码后处理完成。提取的依赖: {required_deps or '无'}。")
        return final_code.strip(), required_deps

    def _install_dependencies(self, deps: Set[str]):
        if not deps:
            logger.info("代码中未发现 # REQUIRE: 声明，跳过依赖安装。")
            return
        deps_to_install = {dep for dep in deps if dep.lower() not in self.standard_libs}
        if not deps_to_install:
            logger.info(f"所有声明的依赖 {list(deps)} 均为标准库，无需安装。")
            return
        
        command = [self.pip_executable, "install", *deps_to_install]
        logger.info(f"执行依赖安装命令: {' '.join(command)}")
        
        result = subprocess.run(
            command,
            cwd=str(self.shared_work_dir),
            capture_output=True,
            text=True,
            timeout=300,
            check=False,
            encoding='utf-8',
        )
        
        if result.returncode != 0:
            error_message = f"依赖安装失败: {result.stderr}"
            logger.error(error_message)
            raise RuntimeError(error_message)
            
        logger.info(f"依赖 {list(deps_to_install)} 安装成功。")

    def _execute_code(self, code_to_execute: str, task_work_dir: str) -> Dict[str, Any]:
        script_name = "generated_script.py"
        code_path = os.path.join(task_work_dir, script_name)
        with open(code_path, "w", encoding="utf-8") as f:
            f.write(code_to_execute)
        
        logger.info(f"最终执行的脚本已保存至: {code_path}")
        
        command = [self.python_executable, script_name]
        logger.info(f"执行代码命令: {' '.join(command)}")
        
        result = subprocess.run(
            command,
            cwd=task_work_dir,
            capture_output=True,
            text=True,
            timeout=300,
            check=False,
            encoding='utf-8',
        )
        
        return {"output": result.stdout, "error": result.stderr, "returncode": result.returncode}

    def _is_direct_command(self, instruction: str) -> bool:
        """
        【新】检查指令是否为直接的 shell 命令。
        """
        # 简单但有效的检查：如果指令以常见命令开头，则认为是直接命令
        common_commands = ['python', 'pip', 'uv', 'bash', 'sh', 'ls', 'rm', 'cp', 'mv', 'mkdir']
        return any(instruction.strip().startswith(cmd) for cmd in common_commands)

    def _execute_direct_command(self, command: str, task_work_dir: str) -> Dict[str, Any]:
        """
        【新】在虚拟环境的上下文中直接执行 shell 命令。
        """
        logger.info(f"检测到直接命令，将在虚拟环境中执行: {command}")
        
        # 为了在虚拟环境中执行，我们构造一个激活并执行的命令
        # 这对于需要虚拟环境上下文的命令（如 pip install）至关重要
        if sys.platform == "win32":
            activate_script = Path(self.python_executable).parent / "activate.bat"
            full_command = f'call "{activate_script}" && {command}'
        else:
            activate_script = Path(self.python_executable).parent / "activate"
            full_command = f'. "{activate_script}" && {command}'

        result = subprocess.run(
            full_command,
            cwd=task_work_dir,
            capture_output=True,
            text=True,
            timeout=300,
            check=False,
            encoding='utf-8',
            shell=True, # shell=True 对于 `.` 和 `&&` 是必需的
            executable='/bin/bash' if sys.platform != "win32" else None
        )
        
        return {"output": result.stdout, "error": result.stderr, "returncode": result.returncode}

    async def run_workflow(self, instruction: str, server_name: str, ctx: Context):
        result = {
            "stages": {
                "code_generation": "pending",
                "dependency_installation": "pending",
                "execution": "pending"
            },
            "code": "", "output": "", "error": "", "work_dir": "", "returncode": -1
        }
        try:
            timestamp = datetime.datetime.now().strftime("task_%Y%m%d_%H%M%S")
            task_work_dir = self.shared_work_dir / timestamp
            task_work_dir.mkdir(exist_ok=True)
            result["work_dir"] = str(task_work_dir)
            await ctx.info(f"任务工作目录已创建: {task_work_dir}")

            # --- 指令嗅探和模式切换 ---
            if self._is_direct_command(instruction):
                await ctx.info("检测到直接命令模式。")
                result["stages"]["code_generation"] = "skipped (direct command)"
                result["stages"]["dependency_installation"] = "skipped (direct command)"
                result["code"] = f"# Direct Command Execution\n{instruction}"

                result["stages"]["execution"] = "pending"
                await ctx.info(f"正在直接执行命令: {instruction}")
                exec_result = self._execute_direct_command(instruction, str(task_work_dir))
                result.update(exec_result)
                
                is_successful = exec_result.get("returncode") == 0
                result["stages"]["execution"] = "success" if is_successful else "failed"
                final_status = "success" if is_successful else "failed"
                
                return {"status": final_status, "result": result, "server": {"name": server_name, "model": "N/A (Direct Command)", "status": "active"}}
            
            # --- 如果不是直接命令，则执行原有的 LLM 工作流 ---
            await ctx.info("进入 LLM 代码生成模式。")
            server_config = self.config.get("mcpServers", {}).get(server_name)
            if not server_config: raise ValueError(f"服务器 '{server_name}' 未配置")
            
            template = server_config["task_templates"]["file_operations"]
            prompt = template["prompt"].format(task=instruction)
            
            await ctx.info(f"正在使用模型 '{server_config['model']}' 生成代码...")
            llm = await self.get_llm(server_config['model'])
            generated_code = await llm.ainvoke(prompt)
            result["stages"]["code_generation"] = "success"
            await ctx.info("代码生成成功。")
            
            pure_code, dependencies = self._post_process_code(generated_code)
            result["code"] = pure_code
            
            result["stages"]["dependency_installation"] = "pending"
            await ctx.info(f"正在检查并安装依赖: {dependencies or '无'}")
            self._install_dependencies(dependencies)
            result["stages"]["dependency_installation"] = "success"
            await ctx.info("所有依赖已就绪。")
            
            result["stages"]["execution"] = "pending"
            await ctx.info("正在执行生成的代码...")
            exec_result = self._execute_code(pure_code, str(task_work_dir))
            
            result.update(exec_result)
            
            is_successful = "任务成功完成" in exec_result.get("output", "")
            result["stages"]["execution"] = "success" if is_successful else "failed"

            final_status = "success" if is_successful else "failed"
            return {"status": final_status, "result": result, "server": {"name": server_name, "model": server_config.get("model"), "status": "active"}}

        except Exception as e:
            current_stage = next((s for s, status in result["stages"].items() if status == "pending"), "unknown")
            result["stages"][current_stage] = "failed"
            error_message = f"在 '{current_stage}' 阶段失败: {e}"
            result["error"] = error_message
            logger.error(error_message, exc_info=True)
            await ctx.error(error_message)
            
            for stage, status in result["stages"].items():
                if status == "pending":
                    result["stages"][stage] = "skipped"

            return {"status": "failed", "result": result, "server": {"name": server_name, "model": self.config.get("mcpServers", {}).get(server_name, {}).get("model"), "status": "error"}}

# --- 单例和工具定义 ---
workflow_executor = TaskWorkflow()

@mcp.tool()
async def execute_natural_command(
    instruction: str = Field(
        description="用户用自然语言下达的、需要在本地计算机上执行的具体任务指令。例如：'在桌面上创建一个名为'todo.txt'的文件' 或 '将/home/user/docs目录下的所有.log文件压缩成一个zip包'。"
    ),
    server: str = Field(default="default", description="要使用的服务器配置名称。"),
    ctx: Context = Field(exclude=True)
) -> CommandOutput:
    """
    当用户需要执行任何与文件、文件夹或系统相关的本地操作时，请使用此工具。
    此工具接收一条自然语言指令，将其转换为可执行的Python代码，并在安全的虚拟环境中运行，然后返回详细的执行结果。
    """
    try:
        await ctx.info(f"收到指令，开始处理: '{instruction}'")
        
        result_dict = await workflow_executor.run_workflow(
            instruction=instruction,
            server_name=server,
            ctx=ctx
        )
        
        await ctx.info("任务流程执行完毕。")
        
        return CommandOutput.model_validate(result_dict)
    except Exception as e:
        await ctx.error(f"执行工具时发生严重错误: {e}")
        
        return CommandOutput(
            status="failed",
            result=ExecutionResult(
                stages=ExecutionStage(
                    code_generation="failed",
                    dependency_installation="skipped",
                    execution="skipped"
                ),
                code="",
                output="",
                error=f"执行工具时发生顶层错误: {e}",
                work_dir="",
                returncode=-1
            ),
            server=ServerInfo(name=server, model=None, status="error")
        )

# --- 服务器启动---
def run():
    """
    服务器主入口函数。
    """
    try:
        subprocess.run(["ollama", "list"], check=True, capture_output=True, text=True)
        logger.info("Ollama 服务已在运行。")
    except (subprocess.CalledProcessError, FileNotFoundError):
        logger.warning("Ollama服务未运行，请确保Ollama已安装并正在运行。")
    
    mcp.run()

if __name__ == "__main__":
    run()

使用方法

保存上面的代码为main.py或其他名字
如果是linux的话，先使用python创建虚拟环境

python -m venv myenv

source相应文件以进入虚拟环境
Windows似乎可以不用创建虚拟环境，但建议创建
pip安装所需要的依赖
进入你的支持mcp的客户端，通过以下json导入这个mcp服务

{
  "mcpServers": {
    "一个名字": {
      "command": "你的路径/.venv/bin/python",
      "args": [
        "你的路径/main.py"
      ]
    }
  }
}

大功告成，可以使用mcp+ollama+python实现的万能工具了！