Как дать ИИ-агенту работать с данными и не потерять контроль: безопасный data-join через MCP, вместо создания DataLake

Это продолжение новых безопасных паттернов по работе с MCP, которые я для себя придумал, которые я описал в статье:

Интересно будет услышать ваше мнение.

Теперь немного истории, проблема копилась в части того, что при развитии продукта постоянно появлялись запросы по аналитике, и все они упирались в решение, «А давайте сделаем DataLake». За больше чем 20 лет я очень много внедрял BI, OLAP, MOLAP и тд. У меня в начале 2000-х был какой-то фетиш на книги Тома Кайта по Ораклу и я вообще думал, что Оракл это лучшая база на свете. Потом были увеличения Дейтом, Кимбалом, 3х уровневой моделью хранилища, 2х, Data Lake, Molap тд. Но щас чего-то немного надоело.

Возникла простая мысль:

«А можно ли налету объединять наборы данных на лету, без необходимости их куда-то перегружать, при этом преобразование будет выполнено в момент обращения к ним. Такое точечный миникуб»

Проблема

Классический запрос от аналитика звучит так:

“Вот clients.csv. Найди тех, у кого не было активности за 30 дней.”

Обычно это значит:

1. ручной выбор product/окружения;

2. ad-hoc SQL в продовой БД;

3. копирование результатов в ноутбук (если нет DataLake, который надо строить);

4. риск перепутать продукт или вытащить лишнее.

Если добавить в эту схему ИИ-агента и просто дать ему “универсальный тул”, риски умножаются:

• агент неявно уходит в чужой продукт;

• грузит слишком много данных “на всякий случай”;

• после ошибки циклически повторяет тот же запрос;

• результат невозможно нормально расследовать в аудите.

Нужен был не “умный SQL”, а безопасный, типизированный, управляемый API для агента. Например:
я хочу понять какие клиенты, у меня не заходили в продукт на этой неделе:
— беру список клиентов например, выгружаю в CSV
— беру сессии по user_id
— пишу в курсор, подготовь список клиентов кто неделю не заходил в сервис и построить воронку по ретеншену. он готовит.

но как? так чтобы еще и данные не скомпрометировать. Так чтобы еще и перс.данные не грузить клиентов. И так чтобы на сервере не осталось перс.данных.

И тут родилась идея, как это сделать.

Идея: не SQL-песочница, а сценарные data-тулы

Часто руководя аналитиками и подразделениями, я видел одну и туже проблему. Люди просто смотрят таблички, без какого-то понятного сценария использования данных. Часто руководители просто просили таблички ради табличек, это было удручающее. Когда ты спрашивал, «А зачем ты это посчитал», в ответ обычно было какое-то мычания. И вот я решил выписать набор сценариев, которые чаще всего спрашивают, и получились набор бизнес-сценариев, каждый из которых мог разложиться на один и тот же набор атомарных тулов. Это по сути очень похоже на SQL, только по смыслу ближе к концепции MCP.

Поскольку часто меня спрашивали про воронки ретеншен, применительно к сущностям products , компании (experiments), клиенты (customers) и данные из яндекс метрики analytics , уменя получились вот такие вот такие атомарные data-тулы, которые могут покрыть много вопросов связанынх с этими сущностями:

1.  products_list | список доступных продуктов | скоуп products:read |

2.  keys_suggest | подсказка join-ключей по колонкам CSV | скоуп analytics:read |

3. customers_lookup | сверка external_id в рамках product_ids | скоупcustomers:read |

4. metrica_events_search | поиск сырых событий Метрики с фильтрами и курсором | скоуп analytics:read

5. funnel_build | многошаговая воронка (до 6 шагов, до 60 дней) | скоуп analytics:read |

6. utm_bind_to_experiment | привязка UTM к эксперименту (с dry-run) | скоуп experiments:write |

У агента при этом нет параметра «выполни произвольную команду». Он может только вызывать заранее определенные сценарии, составленные из data-тулов и прописанные в PlayBook.

Архитектура

Сама архитектура доступа получилась многослойная, поскольку работа с MCP проходит через открытую сеть.

Слой 1. Product isolation как первый инвариант

Основная задача слоя, чтобы агент случай не попал в не нужный продукт. Дополнительно я добавил разграничение доступа к продуктам, у одного аналитика есть доступ к одному продукту, у другого к другому. Поэтому я сделал ролевой сервис AccessService , в рамках, которого проверяется к какому из продуктов у тебя есть доступ, это вынесено в единый resolveProductsOrThrow(). Пример части реализации

async resolveProductsOrThrow(ctx: AccessContext, input: ResolveProductsInput) {  const accessible = await this.getEffectiveAccessibleProducts(ctx);  const requested = input.requestedProductIds;  if (!Array.isArray(requested) || requested.length === 0) {    throw new AnalyticsError('PRODUCT_REQUIRED', 'product_id is required for this tool', {      next_action: 'pick_product',    });  }  const forbidden = requested.filter((id) => !new Set(accessible).has(id));  if (forbidden.length > 0) {    throw new AnalyticsError('PRODUCT_FORBIDDEN', `Access denied for product(s): ${forbidden.join(', ')}`, {      next_action: 'request_product_access_from_admin',    });  }}

Логика простая каждый пользователь регистрируясь в сервисе, может зарегистрировать свой MCP. Для каждого пользователя создается связь между тем, к какому продукту он имеет доступ, а какому нет.

Слой 2. Scope-гейты и явный cross-product

Если у пользователя есть доступ к нескольким продуктам, то я прописываю это в скоупе доступа к продукту

const cross = requested.length > 1;if (cross) {  this.requireScope(ctx, 'analytics:cross_product');}

Иначе — отказ с понятным действием для агента (request_scope_from_user или pick_product). Основная идея тут это объяснить агенту, почему не работает доступ и попросить это у пользователя.

Это важнее, чем кажется. Большинство утечек в аналитике происходят не от «хакеров», а от неявного расширения области выборки, когда мы даем пользователю гораздо больше чем нужно для выполнения его сценария. За 20 лет я видел, как данные используются неэффективно

Слой 3. Машиночитаемые ошибки вместо «текста для человека»

Теперь мы должны настроить обработчик, так чтобы на выходе вместо непонятных ошибок, был понятный транслятор на человеческом языке. Почему? Если агент получает просто строку «доступ запрещен», он часто начинает угадывать дальше, поэтому ему нужна понятная четкая инструкция на выходе. Значит нам нужно унифицировать все доменные ошибки для наших тулов и сценариев

export class AnalyticsError extends Error {  constructor(    public readonly code: AnalyticsErrorCode,    message: string,    public readonly details?: AnalyticsErrorDetails,  ) {    super(message);  }}

Для этого я сделал класс AnalyticsError, который в формате типизированного ответа возвращает в MCP й tool-result с isError: true, error.code и details.next_action (где указывается инструкция чего дальше делать)

на основании инструкции добавляем fallback обработку в агента:

return {  isError: true,  error: payload,  content: [{ type: 'text', text: `[${payload.code}] ${payload.message} ...` }],};

В результате агент не «паникует», а делает следующий корректный шаг: попросить product_id, сузить payload, запросить scope, и т.д. По сути, это такой аналог формирования плана следующих действий, но построенного на архитектурном принципе обработки fallback’ов получился. Мне нравится.

Слой 4. Hard limits вместо доверия к модели

По умолчанию все модели и агенты находятся в 0й зоне доверия, то есть я сразу по умолчанию подразумеваю, что они могут меня например заddos’ить или просто всю память сьесть. Поэтому сразу выставляем лимиты для взаимодействия. Зачем это надо? ну для того, чтобы ваш агент случайно не потащил все данные на свете и грохнул несколько серверов.

Лимиты поставил экспертно

- `customers_lookup`: максимум 500 `external_ids` за вызов;- `metrica_events_search`: лимит страницы до 1000;- `funnel_build`: максимум 6 шагов и окно до 60 дней.Пример из `CustomersLookupService`:

кажется, что их хватает, если пользователи жалуются то просто их улучшаю. Но вроде пока не жалуются:). отдельно я сделал оптимизацию, что при выгрузке большого количества данных, они все равно нарезаются на небольшие кубики

static readonly MAX_LIMIT = 500;const cap = Math.min(CustomersLookupService.MAX_LIMIT, input.limit ?? CustomersLookupService.MAX_LIMIT);const slice = dedup.slice(0, cap);const truncated = dedup.length > cap;

На будущее надо вынести все эти лимиты в админку, но это потом. То есть даже если агент получил CSV на 50k строк, он обязан работать батчами, а не «сливать всё одним запросом».

Слой 5. Аудит как неотключаемый контур

Поскольку запрос идет через интернет, и все MCP у меня обернут в авторизацию, скоуп, и ролевую модель, но все равно данные чувствительные, поэтому всегда надо фиксировать след, кто куда ходил и чего запрашивал. Для этого я сделал отдельный контур ИБ, где фиксируются все события по запрашиванию данных через эти тулы. Он неотключаемый

Каждый вызов аналитического тула пишется в analytics_audit:

• кто вызвал (user_id, agent_id, роль);

• что вызвал (tool_name, event_type, entity);

• по каким продуктам (requested_product_ids, effective_product_ids, cross_product);

• с каким итогом (status, denied_reason, rows_returned).

В MCP сервисе это встроено прямо в пайплайн обработки:

await this.analyticsAuditService.write({  ...auditBase,  entity: 'customers',  effective_product_ids: result?.query?.product_ids ?? [],  cross_product: !!result?.cross_product,  rows_returned: result?.matched?.length ?? 0,  status: 'ok',});

При ошибках AnalyticsError (которые в 1м слое разбирали) пишется denied/error с кодом причины. Это убирает «черную дыру», когда непонятно, что именно агент пытался сделать.

Слой 6. Плейбук в initialize: агенту сразу дается правильный протокол

Для того, чтобы вся эта концепция работала, нужно чтобы агенту объяснили, какие тулы есть, какие вопросы для них задаются. Для этого нужен Playbook, который загружается в агента на этапе инициализации и там мы расписываем что и как. Главное его обязательно проверить самому, а не отдавать на откуп ИИ

Самое важное: В MCP initialize.instructions мы подмешиваем playbook data-join.md только если у токена есть нужные scopes. Ну то есть не всем подряд, когда нет ключей у человека, а ты ему описываешь внутреннее убранство квартиры.

Ключевой принцип playbook-а:

1. Сначала список продуктов — products_list.

2. Потом ключевые вопросы и ответы — keys_suggest/schema. Здесь как раз фиксируется какие join’ы нужно применить (то есть какие таблицы с какими объединяются)

3. Потом целевой принцип работы с тулами — lookup/export.

4. Потом как обрабатывать получаемые ошибки. На любой error.code — читать details.next_action, а не повторять запрос вслепую.

Эта «мягкая оркестрация» резко снижает мусорные вызовы и улучшает устойчивость агентного поведения.

Вот примеры:

Сценарий 1: «Найди неактивных клиентов из CSV»

Рабочий пайплайн процесса:

1. Агент читает локальный CSV (в своей среде, на ноуте)

2. Вызывает products_list, выбирает product_id.

3. Через keys_suggest выбирает join-ключ (client_id -> external_id).

4. Бьет customers_lookup батчами по 500.

5. Локально делает join и фильтр last_activity < now - 30d.

Что важно архитектурно:

• на выходе формируется код, который выполняется в памяти агента

• сервер не исполняет произвольный SQL от агента;

• продуктовая изоляция проверяется на каждом батче;

• данные нигде не сохраняются, агент получает их в памяти, выполняет код in-memory и отдает дальше

• если лимит превышен, агент получает структурированную подсказку «бей на меньшие чанки».

Сценарий 2: «Сделай ретеншен воронку email -> /pricing -> goal»

Для маркетингового анализа, нужно подготовить воронку как пользователи возвращаются:

1. Грузим данные из яндекс метрики metrica_goals_list — валидируем сразу цели goal_id.

2. Считаем воронку (funnel_build ), вычисляем все необходимые шаги для ее построения .

3. При необходимости делаем поиск в событиях ЯМ metrica_events_search , чтобы добавить детализации по клиентам, которые выпали из ретена.

В funnel_build есть важные предохранители:

• ограничение окна в 60 дней;

• запрет слишком общих url_pattern типа /, чтобы избежать ситуации, когда мы просим ретен всего на свете. Только конкретный сценарий

• дедубликация пользователей по умолчанию (через identity_maps);

Это делает воронку и полезной, и предсказуемой с точки зрения качества данных.

Что получилось на практике

Довольно необычный паттерн, базовые аналитические вопросы начали закрываться через MCP в процессе общения с агентом. Сервисом уже пользуются больше 30 человек. И мне не понадобилось строить DataLake, просто в каждом из инстансов, выбираю кусочки данных, формируют программный код, запускаю с ними вычисления и выдаю клиенту.

Появилась появилась повторяемая схема:

• пользователь формулирует вопрос на человеческом языке;

• агент идет по playbook, запрашивает данные, формирует код, чтобы их обработать и вернуть пользователю

• backend строго ограничивает область данных и объем из разных серверов;

• аудит фиксирует каждое действие.

Итог — быстрее цикл аналитики и меньше операционных рисков.

Поресерчил у Антропик и OpenAI такого паттерна я не нашел, поэтому и назвал его data-join-tools. немного похоже на то как MS Access подключаются data-sources, а вы из них собираете ваши запросы. только в этом случае интерфейсом выступает ИИ агент.

Ну и чтобы не казалось малиной, основные ограничения:

1. Некоторые поля/описания в MCP и playbook живут в быстро меняющемся слое; нужен постоянный drift-контроль между docs и регистрацией тулов. С другой стороны это можно поручить другому агенту

2. Ошибки хорошо структурированы, но качество агентного поведения все равно зависит от дисциплины и осознанности пользователя (например, не все агенты одинаково хорошо читают next_action потому что пользователь может написать какую-то ахинею там). нужна защита от дурака

3. Полноценная dataset-механика (серверный upload/join с таблицами, TTL и т.д.) в текущей реализации не созадавалась. Ну не было такой цели. Но я думаю, что этот паттерн можно продолжить и туда. Я начал с data lookup инструментов

В заключении небольшие рекомендации, если вдруг вы также решите сделать такого агента

1. Не давайте агенту произвольный SQL/SSH для аналитики (никаких там select * from все_насвете)

2. Разбивайте тулы по продуктовым доменам и продуктам. Делайте product_id обязательным для data-тулов,

3. Если у вас пользователи имеют доступ к разным продуктам, то проверяйте к каким данным они могут получить доступ в рамках отдельно cross-product на отдельным scope.

4. Возвращайте ошибки в формате error.code + details.next_action, чтобы агент знал что делать дальше

5. Ставьте жесткие лимиты на payload и окна.

6. Аудируйте каждый вызов с requested/effective_product_ids. Завтра придет ИБ, и спросит это че такое, а у вас все логи и все закрыто.

7. Подмешивайте playbook в initialize, чтобы агент работал по протоколу, но обязательно проверяйте скоуп и доступы сначала.

8. Явно документируйте ограничения для агента, чтобы он не галлюцинировал

#ai, #mcp, #data-join-tools