Ваш AI слеп без файлового менеджера
Представьте: у вас есть Claude 4.5 Ultra или GPT-5 Omni — модели, способные анализировать терабайты данных. Но они не видят ваши документы, код, логи. Они — гениальные узники, запертые в клетке из токенов. MCP (Model Context Protocol) создан, чтобы выломать дверь. Но, как показало наше расследование «Кризис Model Context Protocol: 52% удаленных серверов мертвы», полагаться на чужие серверы — путь в никуда. Решение — свой локальный MCP сервер. И сделать его можно на чистом Python, без единой зависимости.
Звучит как вызов? На самом деле это проще, чем кажется. В этой статье я покажу, как написать zero-dependency MCP сервер, который даст вашему AI-агенту доступ к файловой системе, и при этом не принесёт в проект ни одного внешнего пакета. Только стандартная библиотека Python, только хардкор.
Почему zero-dependency, а не pip install mcp?
Библиотеки для MCP есть. Но:
- Они часто обновляются — ловить баги после `pip install --upgrade`? Не хочу.
- Каждая зависимость — это поверхность для атаки. Помните историю с
event-stream? То-то же. - Размер контейнера. Для локального агента тащить 50 МБ зависимостей ради чтения файла — жирно.
Наш сервер будет весить ровно столько, сколько весит сам Python. И работать будет в любом окружении — хоть на Raspberry Pi Zero, хоть в изолированном Docker.
Архитектура: MCP — это просто JSON-RPC по трубе
Забудьте про магию. MCP — это клиент-сервер, где клиент (AI-хост: Claude Desktop, Cursor, VS Code) шлёт JSON-RPC 2.0 сообщения, а сервер отвечает результатами. Есть два основных транспорта:
- stdio — сервер запускается как подпроцесс, общается через stdin/stdout. Идеально для локальных агентов.
- HTTP + SSE — сервер работает как веб-сервер, клиент подключается по SSE. Для удалённого доступа или когда нужно несколько подключений.
Мы реализуем оба. Но начнём с stdio — это проще и безопаснее.
Шаг 1. JSON-RPC 2.0 на коленке
В центре любого MCP сервера — диспетчер сообщений. Наш будет читать из stdin, парсить JSON, выполнять метод и писать ответ в stdout.
import sys, json, os, traceback
class MCPDispatcher:
def __init__(self):
self.handlers = {}
def register(self, method, handler):
self.handlers[method] = handler
def handle(self, message):
if 'method' not in message:
return self._error(message.get('id'), -32600, 'Invalid Request')
method = message['method']
handler = self.handlers.get(method)
if not handler:
return self._error(message['id'], -32601, f'Method not found: {method}')
try:
result = handler(message.get('params', {}))
return {'jsonrpc': '2.0', 'id': message.get('id'), 'result': result}
except Exception as e:
return self._error(message['id'], -32603, str(e))
def _error(self, id, code, message):
return {'jsonrpc': '2.0', 'id': id, 'error': {'code': code, 'message': message}}
def run_stdio(self):
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError:
resp = self._error(None, -32700, 'Parse error')
print(json.dumps(resp), flush=True)
continue
resp = self.handle(msg)
print(json.dumps(resp), flush=True)Этот код — базовый каркас. Он слушает stdin, парсит построчно, вызывает нужный обработчик и печатает ответ. Обработка ошибок по спецификации JSON-RPC.
Важно! В MCP сообщения могут быть уведомлениями (без id). Их не нужно обрабатывать в данном примере, но для полной реализации следует различать request и notification (id отсутствует).
Шаг 2. Инструменты файловой системы
MCP сервер публикует список инструментов. Клиент (AI) может их вызывать. Мы реализуем четыре базовых:
list_directory— получить содержимое папкиread_file— прочитать файлwrite_file— записать файлsearch_files— поиск по маске
Добавим безопасность: сервер будет работать только в рамках одной разрешённой директории (sandbox root).
import pathlib
class FileSystemTools:
def __init__(self, sandbox_root):
self.root = pathlib.Path(sandbox_root).resolve()
if not self.root.exists():
os.makedirs(self.root, exist_ok=True)
def _resolve(self, path):
# Абсолютный путь внутри sandbox
full = (self.root / path).resolve()
# Проверяем, что full начинается с self.root (защита от path traversal)
if not str(full).startswith(str(self.root)):
raise PermissionError('Access denied')
return full
def list_directory(self, params):
path = params.get('path', '.')
full = self._resolve(path)
if not full.is_dir():
raise ValueError(f'{full} is not a directory')
entries = []
for child in full.iterdir():
entries.append({
'name': child.name,
'type': 'directory' if child.is_dir() else 'file',
'size': child.stat().st_size if child.is_file() else 0
})
return {'entries': entries, 'path': str(path)}
def read_file(self, params):
path = params.get('path')
full = self._resolve(path)
if not full.is_file():
raise FileNotFoundError(f'{full} not found')
# Безопасный лимит на размер
max_size = params.get('max_size', 1024*1024) # 1 MB по умолчанию
if full.stat().st_size > max_size:
raise ValueError('File too large')
content = full.read_text(encoding='utf-8')
return {'content': content, 'path': str(path), 'size': len(content)}
def write_file(self, params):
path = params.get('path')
content = params.get('content')
full = self._resolve(path)
# Дополнительная проверка: можно ли писать в этой директории?
full.parent.mkdir(parents=True, exist_ok=True)
full.write_text(content, encoding='utf-8')
return {'success': True, 'path': str(path)}
def search_files(self, params):
pattern = params.get('pattern', '*')
root = params.get('root', '.')
full_root = self._resolve(root)
results = []
for p in full_root.rglob(pattern):
if p.is_file():
results.append(str(p.relative_to(self.root)))
return {'results': results[:100]}Безопасность — не опция. Функция _resolve проверяет, что итоговый путь находится внутри sandbox. Это защищает от атак типа ../../etc/passwd. Также мы ограничиваем размер читаемого файла — AI может попросить прочитать гигабайтный лог, но мы вежливо откажем.
Шаг 3. Регистрируем инструменты и запускаем
if __name__ == '__main__':
# sandbox_root можно передать через переменную окружения
sandbox = os.environ.get('MCP_SANDBOX', '.')
tools = FileSystemTools(sandbox)
dispatcher = MCPDispatcher()
dispatcher.register('list_directory', tools.list_directory)
dispatcher.register('read_file', tools.read_file)
dispatcher.register('write_file', tools.write_file)
dispatcher.register('search_files', tools.search_files)
# Публикуем список инструментов (обязательно для MCP)
# В реальном MCP это делается через initialize + list_tools
# Упростим: добавим метод list_tools
dispatcher.register('list_tools', lambda p: {
'tools': [
{'name': 'list_directory', 'description': 'List directory contents',
'inputSchema': {'type': 'object', 'properties': {
'path': {'type': 'string', 'default': '.'}
}}},
# ... аналогично для других
]
})
dispatcher.register('initialize', lambda p: {
'protocolVersion': '2024-11-05',
'capabilities': {'tools': {}}
})
dispatcher.run_stdio()Теперь, если запустить этот скрипт, любой MCP-клиент сможет к нему подключиться. Например, в конфиге Claude Desktop:
{
"mcpServers": {
"filesystem": {
"command": "python3",
"args": ["/path/to/your/server.py"]
}
}
}После этого ваш Claude сможет читать и писать файлы в указанной директории. Просто и без единого pip install.
Шаг 4. HTTP + SSE транспорт для удалённого доступа
Локальный stdio удобен, но иногда хочется, чтобы сервер был доступен по сети. Например, чтобы подключить его к удалённому агенту. Реализуем HTTP сервер с SSE.
import http.server
import socketserver
import re
class SSEMCPHandler(http.server.BaseHTTPRequestHandler):
dispatcher = None
def do_POST(self):
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length).decode('utf-8')
try:
msg = json.loads(body)
except:
self.send_response(400)
self.end_headers()
self.wfile.write(b'Invalid JSON')
return
resp = self.__class__.dispatcher.handle(msg)
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(resp).encode())
def do_GET(self):
# SSE endpoint
if self.path == '/sse':
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self.end_headers()
# В реальном сервере нужно слушать канал и отправлять события
# Здесь заглушка
self.wfile.write(b'data: {"event": "connected"}\n\n')
self.wfile.flush()
# Бесконечный цикл держит соединение открытым
while True:
# В реальности тут был бы asyncio или threading
time.sleep(1)
else:
self.send_response(404)
self.end_headers()
def run_http(host='0.0.0.0', port=9090, dispatcher=None):
SSEMCPHandler.dispatcher = dispatcher
server = http.server.HTTPServer((host, port), SSEMCPHandler)
print(f'MCP HTTP server listening on {host}:{port}')
server.serve_forever()Осторожно: Этот HTTP сервер — синхронный и однопоточный. Для продакшена лучше использовать asyncio или народный ThreadingHTTPServer. Но для локальной разработки и отладки — вполне.
Тестируем сервер без клиента
Чтобы проверить, что всё работает, можно имитировать MCP-клиент через curl (для HTTP) или через echo (для stdio).
# Для stdio сервера:
echo '{"jsonrpc":"2.0","id":1,"method":"list_tools"}' | python server.py
# Должен вывести список инструментов# Для HTTP сервера:
curl -X POST http://localhost:9090 \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"list_directory","params":{"path":"."}}'Если видите JSON с результатом — всё работает. Готово к подключению к реальному AI.
Ошибки и грабли, на которые я наступал
Вот что пошло не так у меня, и как это исправить:
| Проблема | Причина | Решение |
|---|---|---|
| AI не видит инструменты | Не отправлен метод list_tools | Убедитесь, что на initialize возвращается capabilities.tools |
| Ошибка парсинга при больших файлах | Ответ содержит >1 MB текста — клиент обрывает | Добавьте пагинацию или чанкинг |
| Path traversal сработал | Не использовал .resolve() | Всегда резолвьте и проверяйте префикс |
| Сервер завис | Бесконечный цикл в SSE | Добавьте таймаут или используйте asyncio |
Безопасность: как НЕ надо делать
Я видел серверы, которые выставляют / как sandbox root. Это преступление. Вот минимальные правила:
- Никогда не давайте доступ к
/,C:\или~. - Всегда проверяйте, что итоговый путь начинается с разрешённого корня (через
os.path.realpathиstartswith). - Ограничьте размер записываемых файлов — AI может заспамить ваш диск.
- Если сервер на HTTP — используйте API ключ или вообще только локальный хост.
Подробнее про угрозы читайте в материале «Как защитить AI-агентов от prompt injection».
Что дальше?
Zero-dependency MCP сервер — это база. Но вы можете нарастить на него мясо: добавить асинхронность, воркфлоу (как в MCP Chat Studio v2), поддержку нескольких sandbox-зон, интеграцию с шифрованием. Главное — вы теперь не зависите от сторонних библиотек и чужих серверов.
И ещё: не забудьте поделиться своим сервером. Возможно, именно ваш код станет стандартом для локальных агентов. Как говорят в Open Source: «Если это не задокументировано — этого не существует». А если это не zero-dependency — это не ваш сервер.