mcp-python/tools.py

import os
import sys
import json
import asyncio
import platform
import subprocess
from typing import List, Dict, Any
from mcp.types import Tool

# --- ОПРЕДЕЛЕНИЕ ИНСТРУМЕНТОВ ---

def get_tool_definitions() -> List[Tool]:
    return [
        Tool(
            name="run_script",
            description="Запускает Python скрипт в указанной директории.",
            inputSchema={
                "type": "object",
                "properties": {
                    "script_path": {"type": "string", "description": "Путь к скрипту (например, main.py)"},
                    "working_dir": {"type": "string", "description": "Рабочая директория"}
                },
                "required": ["script_path"]
            }
        ),
        Tool(
            name="install_package",
            description="Устанавливает пакет через pip.",
            inputSchema={
                "type": "object",
                "properties": {
                    "package_name": {"type": "string", "description": "Имя пакета"},
                    "working_dir": {"type": "string", "description": "Директория проекта (опционально)"}
                },
                "required": ["package_name"]
            }
        ),
        Tool(
            name="list_files",
            description="Выводит список файлов и папок.",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Путь к папке"}
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="read_file",
            description="Читает содержимое текстового файла.",
            inputSchema={
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "Путь к файлу"}
                },
                "required": ["file_path"]
            }
        ),
        Tool(
            name="get_env_info",
            description="Информация о системе и Python окружении.",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

# --- ЛОГИКА ВЫПОЛНЕНИЯ ---

async def execute_tool(name: str, arguments: Dict[str, Any]) -> str:
    """Маршрутизатор вызовов."""
    if name == "run_script":
        return await _run_script(arguments.get("script_path"), arguments.get("working_dir"))
    elif name == "install_package":
        return await _install_package(arguments.get("package_name"), arguments.get("working_dir"))
    elif name == "list_files":
        return await _list_files(arguments.get("path", "."))
    elif name == "read_file":
        return await _read_file(arguments.get("file_path"))
    elif name == "get_env_info":
        return await _get_env_info()
    else:
        return f"Ошибка: Инструмент '{name}' не найден."

# --- ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ---

async def _run_command(cmd: List[str], cwd: str = None) -> str:
    """Асинхронный запуск команды."""
    try:
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=cwd
        )
        stdout, stderr = await process.communicate()
        
        result = f"STDOUT:\n{stdout.decode('utf-8', errors='replace')}"
        if stderr:
            result += f"\nSTDERR:\n{stderr.decode('utf-8', errors='replace')}"
        
        if process.returncode != 0:
            result += f"\n[Код завершения: {process.returncode}]"
        return result
    except Exception as e:
        return f"Ошибка выполнения команды: {e}"

async def _run_script(script_path: str, working_dir: str = None) -> str:
    if not working_dir:
        working_dir = os.path.dirname(script_path) or os.getcwd()
    return await _run_command([sys.executable, script_path], cwd=working_dir)

async def _install_package(package_name: str, working_dir: str = None) -> str:
    cmd = [sys.executable, "-m", "pip", "install", package_name]
    return await _run_command(cmd, cwd=working_dir)

async def _list_files(path: str) -> str:
    try:
        if not os.path.exists(path):
            return f"Путь не найден: {path}"
        items = os.listdir(path)
        files = [i for i in items if os.path.isfile(os.path.join(path, i))]
        dirs = [i + "/" for i in items if os.path.isdir(os.path.join(path, i))]
        return "Папки:\n" + "\n".join(sorted(dirs)) + "\n\nФайлы:\n" + "\n".join(sorted(files))
    except Exception as e:
        return f"Ошибка чтения: {e}"

async def _read_file(file_path: str) -> str:
    try:
        if not os.path.exists(file_path):
            return f"Файл не найден: {file_path}"
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Ошибка чтения: {e}"

async def _get_env_info() -> str:
    info = {
        "os": platform.system(),
        "python": platform.python_version(),
        "cwd": os.getcwd()
    }
    return json.dumps(info, indent=2, ensure_ascii=False)