server.py•3.12 kB
import asyncio
import os
import mcp.types as types
from mcp.server.models import InitializationOptions
import json
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
import mcp.server.stdio
from datetime import datetime
server = Server("file-finder-mcp")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
Описывает доступные инструменты.
"""
return [
types.Tool(
name="search-files",
description="Поиск файлов по фрагменту пути",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Фрагмент пути для поиска"},
"directory": {"type": "string", "description": "Каталог для поиска", "default": "/"},
},
"required": ["query"],
},
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent]:
"""
Обрабатывает выполнение инструментов.
"""
if name != "search-files":
raise ValueError(f"Неизвестный инструмент: {name}")
if not arguments or "query" not in arguments:
raise ValueError("Отсутствует обязательный параметр 'query'")
query = arguments["query"]
directory = arguments.get("directory", "/")
result = search_files(query, directory)
return [
types.TextContent(
type="text",
text=result,
)
]
def search_files(query: str, directory: str) -> str:
"""
Выполняет поиск файлов в заданном каталоге.
Возвращает результат в формате JSON.
"""
matches = []
for root, _, files in os.walk(directory):
for file in files:
if query in file:
file_path = os.path.join(root, file)
try:
stats = os.stat(file_path)
matches.append({
"name": file,
"path": file_path,
"size": stats.st_size,
"created": datetime.fromtimestamp(stats.st_ctime).isoformat(),
})
except Exception:
continue
return json.dumps({"files": matches}, indent=2)
async def main():
# Run the server using stdin/stdout streams
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="file-finder-mcp",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)