Как создать MCP-сервер для интерактивного извлечения текста из PDF в Amazon S3

Допустим, у вас в S3 лежат десятки тысяч PDF-контрактов, инвойсов или engineering docs. И вы хотите, чтобы ваша LLM могла по запросу залезть в любой файл, вытащить текст и ответить на вопрос. Простая мысль: взять PyPDF2, скачать файл из S3, распарсить — и вуаля? Нет, не вуаля.

PyPDF2 ломается на сканах, pdfminer.six по кодировкам, а если PDF — результат сканирования с картинками, то вы ничего не получите без распознавания. Amazon Textract решает проблему, но напрямую дергать API из каждого вызова LLM — дорого и медленно. Нужен прослойка: MCP-сервер, который кеширует результаты, управляет пагинацией и дает LLM чистый текст.

Я покажу, как собрать такой сервер за вечер. Код на Python, деплой на AWS Lambda, асинхронная обработка через Textract, кеш в DynamoDB. Поехали.

Почему MCP, а не прямой API?

MCP протокол — это USB-порт для ИИ. Вместо того чтобы вшивать AWS SDK в каждую функцию агента, вы выносите работу с PDF в отдельный инструмент. LLM отправляет запрос: «извлеки текст из документа reports/2026-03-01.pdf» — сервер сам скачивает, вызывает Textract (если нет кеша), возвращает строки. Результат: чистый контракт, без зависимостей, без тормозов.

Прямой вызов Textract из кода LLM — антипаттерн: каждый токен стоит денег, а время ожидания ответа 2-3 секунды блокирует агента. MCP-сервер работает асинхронно: отправляет задачу в Textract, возвращает JobId, а потом LLM дергает статус. Это как разница между синхронным get_object и асинхронным StartDocumentTextDetection.

Архитектура

Как НЕ надо делать — типичные грабли

Многие пишут MCP-сервер, который синхронно вызывает Textract через detect_document_text (для одного изображения) — и удивляются, что на 10-страничном PDF прилетает ошибка LimitExceededException. Textract имеет квоту 2 запроса в секунду на аккаунт. При синхронном вызове вы блокируетесь на 5-10 секунд, потом повтор — и снова лимит.

Правильный путь: StartDocumentTextDetection -> сохраняем JobId -> возвращаем LLM статус pending -> LLM опрашивает через GetDocumentTextDetection. Но в MCP это неудобно: протокол предполагает один ответ на запрос. Хитрость: запускаем Textract асинхронно, а MCP-сервер ждет результат с помощью EventBridge или простого sleep+loop.

Пошаговая сборка

1 Подготовка окружения и IAM

Создайте политику с доступом к S3 (только ваш bucket), Textract (полный доступ), DynamoDB (GetItem, PutItem, UpdateItem). Прицепите к роли, которую будете использовать в Fargate.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "textract:StartDocumentTextDetection",
        "textract:GetDocumentTextDetection"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:GetObjectVersion"
      ],
      "Resource": "arn:aws:s3:::your-bucket/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123456:table/pdf-cache"
    }
  ]
}

2 Пишем MCP-сервер

Используем библиотеку mcp (FastMCP). Сервер будет слушать один инструмент: extract_pdf_text с параметрами bucket и key.

import asyncio
import boto3
import hashlib
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("pdf-extractor")
textract = boto3.client("textract", region_name="us-east-1")
s3 = boto3.client("s3")
ddb = boto3.resource("dynamodb").Table("pdf-cache")

@mcp.tool()
def extract_pdf_text(bucket: str, key: str) -> str:
    cache_key = hashlib.sha256(f"{bucket}/{key}".encode()).hexdigest()

    # 1. Проверяем кеш
    cached = ddb.get_item(Key={"cache_key": cache_key})
    if "Item" in cached:
        return cached["Item"]["text"]

    # 2. Запускаем Textract
    response = textract.start_document_text_detection(
        DocumentLocation={
            "S3Object": {
                "Bucket": bucket,
                "Name": key
            }
        }
    )
    job_id = response["JobId"]

    # 3. Ожидаем завершения (максимум 5 минут, шаг 2 сек)
    text = ""
    for _ in range(150):
        result = textract.get_document_text_detection(JobId=job_id)
        if result["JobStatus"] == "SUCCEEDED":
            lines = [b["Text"] for b in result["Blocks"] if b["BlockType"] == "LINE"]
            text = "\n".join(lines)
            break
        elif result["JobStatus"] == "FAILED":
            raise Exception(f"Textract failed: {result.get('StatusMessage', 'unknown')}")
        asyncio.sleep(2)
    else:
        raise TimeoutError("Textract job did not complete in 5 minutes")

    # 4. Сохраняем в кеш с TTL 7 дней
    ddb.put_item(Item={
        "cache_key": cache_key,
        "text": text,
        "ttl": int(time.time()) + 604800
    })
    return text

if __name__ == "__main__":
    mcp.run()

import time
# ...
@mcp.tool()
def extract_pdf_text(bucket: str, key: str) -> str:
    # ...
    for _ in range(150):
        result = textract.get_document_text_detection(JobId=job_id)
        if result["JobStatus"] == "SUCCEEDED":
            # ...
            break
        time.sleep(2)

3 Деплой и тестирование

Я собираю Docker-образ и пулю в ECR, затем запускаю как задачу Fargate с публичной сетью. Вместо этого можно использовать MCP Chat Studio — там есть встроенный клиент для тестирования инструментов. Просто подключаетесь к вашему серверу, дергаете extract_pdf_text, смотрите результат.

Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install mcp boto3
COPY server.py .
CMD ["python", "server.py"]

Не забудьте передать переменные окружения AWS_REGION и роль IAM через Task Role.

Сравнение: MCP vs прямой Textract

В статье про обработку 4700 инженерных PDF похожий подход — но там использовали батчевую обработку без MCP. С MCP вы получаете интерактивность: агент сам решает, какой PDF открыть прямо во время диалога. Это меняет сценарий с batch на on-demand.

Грабли, на которых я обжегся

• TTL в DynamoDB не удаляет элементы мгновенно. Ставите TTL = 604800, но элемент может жить до 48 часов дольше. Для кеша это нормально, но не используйте TTL для инвалидации sensitive данных.
• Textract путает страницы для PDF с изображениями. Если PDF создан из сканов, Textract возвращает блоки в произвольном порядке. Сортируйте по PageNumber из Block.
• Квоты Textract: 2 запроса в секунду для StartDocumentTextDetection. Если вы запускаете сервер и к нему стучатся несколько LLM, введите очередь (SQS) или rate limiting.
• Asyncio и boto3 несовместимы. boto3 блокирует event loop. Используйте boto3.client(...) внутри executor или переходите на aioboto3.