dinlo
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
import sys
|
|
import os
|
|
import logging
|
|
import asyncio
|
|
from typing import List, Dict, Any, Sequence
|
|
|
|
# MCP Imports
|
|
from mcp.server import Server
|
|
from mcp.server.stdio import stdio_server
|
|
from mcp.types import Tool, TextContent
|
|
|
|
# Import tools from local file
|
|
import tools
|
|
|
|
# Настройка логирования
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
|
logger = logging.getLogger("mcp-python-server")
|
|
|
|
# Инициализация сервера
|
|
app = Server("mcp-python-manager")
|
|
|
|
# --- Регистрация инструментов ---
|
|
|
|
@app.list_tools()
|
|
async def list_tools() -> List[Tool]:
|
|
"""Возвращает список доступных инструментов."""
|
|
return tools.get_tool_definitions()
|
|
|
|
@app.call_tool()
|
|
async def call_tool(name: str, arguments: Dict[str, Any]) -> Sequence[TextContent]:
|
|
"""Вызывает конкретный инструмент."""
|
|
logger.info(f"Вызов инструмента: {name}")
|
|
try:
|
|
# Выполняем логику из tools.py
|
|
result_text = await tools.execute_tool(name, arguments)
|
|
return [TextContent(type="text", text=str(result_text))]
|
|
except Exception as e:
|
|
logger.error(f"Ошибка в инструменте {name}: {e}")
|
|
return [TextContent(type="text", text=f"Ошибка выполнения: {str(e)}")]
|
|
|
|
# --- Запуск сервера ---
|
|
|
|
async def main():
|
|
"""Основная функция запуска."""
|
|
logger.info("Запуск MCP сервера на stdio...")
|
|
# Запуск сервера через стандартный ввод/вывод
|
|
async with stdio_server() as streams:
|
|
await app.run(*streams, app.create_initialization_options())
|
|
|
|
def run_server():
|
|
"""Точка входа для uvx."""
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
logger.info("Сервер остановлен пользователем.")
|
|
except Exception as e:
|
|
logger.critical(f"Критическая ошибка сервера: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
run_server() |