Как подключить ИИ-ассистента Альбато к вашему ИИ-агенту по протоколу A2A

Эта инструкция предназначена для разработчиков, которые встраивают Альбато в свой продукт в рамках вайт-лейбл интеграции. Вы узнаете, как подключить своего ИИ-агента к ИИ-ассистенту Альбато по открытому протоколу A2A (Agent2Agent, версия 1.0). После подключения ваш агент сможет поручать ИИ-ассистенту создание автоматизаций, а вы сможете показывать результат пользователю в интерфейсе своего продукта.

1. Что такое ИИ-ассистент (Copilot) по A2A

ИИ-ассистент Альбато (Copilot) — это агент, который от имени пользователя Альбато:

• собирает, настраивает, проверяет и запускает автоматизации: выбирает приложения, триггеры и действия, подключения, сопоставляет поля;
• ищет по каталогу Альбато приложения, их триггеры и действия, показывает подключения пользователя;
• разбирает существующие связки и объясняет, почему они не работают.

По A2A ваш агент отправляет ИИ-ассистенту задачу обычным текстом на языке пользователя, а в ответ получает поток событий: что ИИ-ассистент делает сейчас, какие вопросы у него возникли и итоговый ответ. Технически это JSON-RPC 2.0 поверх HTTPS, поток событий приходит как Server-Sent Events (SSE). Описание протокола и инструменты для разработки доступны на официальном сайте A2A. Для Python используется пакет a2a-sdk.

JSON-RPC 2.0 задаёт формат запросов и ответов. SSE (Server-Sent Events) позволяет серверу передавать события по мере выполнения задачи. DataPart — часть сообщения со структурированными данными.

Основные понятия:

2. Что вы получите от менеджера Альбато

1. Адрес ИИ-ассистента — базовый URL, далее в тексте <COPILOT_URL>.
2. Партнёрский токен вида cpa2a_… — секрет, который идентифицирует вашего агента. Один на партнёра. Храните его на сервере, как любой API-ключ, и никогда не передавайте в браузер. Если токен утёк, напишите менеджеру: его перевыпустят, старый перестанет работать сразу.

3. Авторизация: два токена в каждом запросе

В каждый запрос к <COPILOT_URL>/a2a добавьте два заголовка:

И обязательный служебный заголовок протокола: A2A-Version: 1.0 (SDK добавляет его автоматически; при ручных запросах добавьте вручную).

Необязательно: если пользователь работает в рабочем пространстве Альбато Teams, добавьте заголовки WorkspaceId и SpaceId — те же, что ваш интерфейс отправляет в API Альбато.

4. Как получить токен пользователя и передать его ИИ-ассистенту

ИИ-ассистент принимает тот же токен сессии Альбато, который вы уже используете, чтобы показать пользователю интерфейс Альбато внутри своего продукта: session_token, который вы передаёте в интерфейс Альбато при открытии (в браузере он затем хранится в cookie authToken_<среда>). Отдельный токен пользователя для ИИ-ассистента не нужен.

Рекомендуемая схема:

POST <COPILOT_URL>/a2a
X-Copilot-Partner-Token: cpa2a_…
Authorization: Bearer <session_token пользователя>
A2A-Version: 1.0

В этой схеме бэкенд — серверная часть вашего продукта, а фронтенд — интерфейс, который работает в браузере пользователя.

Правила:

• A2A-клиент работает на вашем сервере. Партнёрский токен нельзя передавать в браузер. Если ваш агент запускается на фронтенде, используйте между ним и ИИ-ассистентом свой прокси-сервер, который добавит партнёрский токен.
• Токен пользователя хранит ваш бэкенд, а не страница. Ваш фронтенд авторизуется в вашем API как обычно, а токен Альбато бэкенд подставляет сам. Так токен не попадает в JavaScript и логи браузера. Если по архитектуре вы всё же передаёте его с фронтенда, делайте это только в теле или заголовке запроса к вашему API по HTTPS и не добавляйте в URL.
• Срок жизни токена равен сроку сессии пользователя в Альбато. Если ИИ-ассистент ответил 401 на пользовательский токен, получите новый токен тем же способом, каким получаете его для интерфейса Альбато, и повторите запрос.
• Один токен пользователя — один пользователь. Не отправляйте задачи одного пользователя с токеном другого: ИИ-ассистент будет действовать в чужом аккаунте.

5. Как отправить задачу и обработать ответ

Шаг 1. Получите карточку агента

curl -s <COPILOT_URL>/.well-known/agent-card.json

Карточка публичная. Из неё нужен адрес JSON-RPC (supportedInterfaces[0].url, это <COPILOT_URL>/a2a) и список навыков. SDK получает эти данные автоматически.

Шаг 2. Отправьте задачу

Рекомендуем потоковый метод SendStreamingMessage: вы сразу видите прогресс, и снижается риск закрытия соединения по таймауту балансировщика нагрузки. Блокирующий SendMessage держит HTTP-соединение до конца задачи (это может занять минуты) и подходит только для коротких запросов.

curl -N -s -X POST <COPILOT_URL>/a2a \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -H "A2A-Version: 1.0" \
  -H "X-Copilot-Partner-Token: $COPILOT_PARTNER_TOKEN" \
  -H "Authorization: Bearer $ALBATO_USER_TOKEN" \
  -d '{
    "jsonrpc": "2.0", "id": "1", "method": "SendStreamingMessage",
    "params": {"message": {"messageId": "m-1", "role": "ROLE_USER",
                           "parts": [{"text": "Какие триггеры есть у Telegram?"}]}}
  }'

messageId — любой уникальный идентификатор с вашей стороны (например, UUID).

Шаг 3. Обработайте поток событий

Каждое SSE-событие — строка data: {...} с JSON-RPC-ответом. В поле result находится одно из следующих событий:

Поток заканчивается вместе с завершением задачи или на состоянии INPUT_REQUIRED.

Шаг 4. Передайте ответы на вопросы ИИ-ассистента (INPUT_REQUIRED)

ИИ-ассистент часто уточняет: какое действие выбрать, какое подключение использовать, что именно нужно сделать. Тогда задача переходит в TASK_STATE_INPUT_REQUIRED, а в status.message передаются:

• текст вопроса — parts[0].text;
• DataPart с деталями — parts[1].data:
  • {"type": "question"} — свободный вопрос, ответьте текстом;
  • {"type": "hitl_request", "hitl_type": "select" | "add_connection" | "connection_actions" | …, "prompt": "…", "options": [{"option_id": "…", "label": "…"}], "connection_url": "…"?} — выбор из вариантов.

Покажите вопрос и варианты пользователю и отправьте его ответ новым сообщением с теми же taskId и contextId:

{"jsonrpc": "2.0", "id": "2", "method": "SendStreamingMessage",
 "params": {"message": {"messageId": "m-2", "role": "ROLE_USER",
                        "taskId": "<taskId>", "contextId": "<contextId>",
                        "parts": [{"text": "My Telegram bot"}]}}}

Чтобы выбрать вариант, передайте option_id, его label (регистр не учитывается) или отдельную часть {"data": {"selected_option_id": "cred_1"}}. Если ответ не совпал ни с одним вариантом, задача останется в INPUT_REQUIRED, а в сообщении будет подсказка со списком вариантов.

Шаг 5. Продолжите разговор

• Следующее сообщение в тот же разговор: передайте contextId из первого ответа (без taskId). ИИ-ассистент помнит контекст: найденные приложения, созданную автоматизацию, предыдущие ответы.
• Новое сообщение без taskId, пока предыдущая задача не завершена: текущая задача прерывается, неотвеченный вопрос отменяется, начинается новая задача.
• Отменить задачу: {"method": "CancelTask", "params": {"id": "<taskId>"}}.
• Узнать состояние: {"method": "GetTask", "params": {"id": "<taskId>"}}.
• Сообщение в завершённую задачу отклоняется — начните новую с тем же contextId.

6. Пример на Python (a2a-sdk)

pip install "a2a-sdk>=1.1" httpx

import asyncio
import os

import httpx
from a2a.client import ClientConfig, create_client
from a2a.helpers import (
    get_data_parts, get_stream_response_text,
    new_data_part, new_message, new_text_part,
)
from a2a.types import Role, SendMessageRequest, TaskState

COPILOT_URL = os.environ["COPILOT_URL"]              # прислал менеджер
PARTNER_TOKEN = os.environ["COPILOT_PARTNER_TOKEN"]  # секрет сервера


async def ask_copilot(user_token, text, *, context_id=None, task_id=None,
                      selected_option_id=None):
    http = httpx.AsyncClient(
        headers={
            "X-Copilot-Partner-Token": PARTNER_TOKEN,
            "Authorization": f"Bearer {user_token}",
        },
        timeout=httpx.Timeout(600, connect=10),
    )
    config = ClientConfig(streaming=True, httpx_client=http)
    client = await create_client(COPILOT_URL, client_config=config)

    parts = [new_text_part(text)] if text else []
    if selected_option_id:
        parts.append(new_data_part({"selected_option_id": selected_option_id}))
    message = new_message(parts, context_id=context_id, task_id=task_id,
                          role=Role.ROLE_USER)

    state, answer, question = None, "", None
    async for chunk in client.send_message(SendMessageRequest(message=message)):
        if chunk.HasField("task"):
            task_id, context_id = chunk.task.id, chunk.task.context_id
        elif chunk.HasField("status_update"):
            status = chunk.status_update.status
            state = TaskState.Name(status.state)
            if state == "TASK_STATE_WORKING":
                progress = get_stream_response_text(chunk)
                if progress:
                    print("[copilot]", progress)  # прогресс
            elif state == "TASK_STATE_INPUT_REQUIRED":
                datas = get_data_parts(status.message.parts)
                question = {
                    "text": get_stream_response_text(chunk),
                    "data": datas[0] if datas else None,
                }
        elif chunk.HasField("artifact_update"):
            answer += get_stream_response_text(chunk)

    await client.close()
    await http.aclose()
    return {"state": state, "answer": answer, "question": question,
            "task_id": task_id, "context_id": context_id}


async def main():
    user_token = "<session_token пользователя Albato>"
    r = await ask_copilot(user_token, "Сделай связку: новый лид в amoCRM "
                                      "→ сообщение в Telegram")
    while r["state"] == "TASK_STATE_INPUT_REQUIRED":
        print("Copilot спрашивает:", r["question"]["text"])
        for opt in (r["question"]["data"] or {}).get("options", []):
            print("  -", opt["option_id"], ":", opt["label"])
        reply = input("Ответ пользователя: ")
        r = await ask_copilot(user_token, reply,
                              context_id=r["context_id"], task_id=r["task_id"])
    print(r["state"], r["answer"])


asyncio.run(main())

7. Ошибки

HTTP-ошибки приходят до JSON-RPC:

Ошибки JSON-RPC приходят с HTTP 200 в поле error:

Ошибки самого ИИ-ассистента (нет доступа к API Альбато, исчерпан лимит, сбой агента) приходят как TASK_STATE_FAILED с текстом для пользователя.

8. Ограничения и советы

• Одна задача выполняется до 15 минут; затем она прерывается и возвращается как FAILED.
• В одном разговоре (contextId) задачи идут последовательно: дождитесь завершения или INPUT_REQUIRED прежде чем отправлять следующее сообщение.
• Пишите ИИ-ассистенту на языке пользователя — ответ придёт на том же языке.
• Чаты A2A-сессий можно будет посмотреть в партнёрском кабинете Альбато.
• Не записывайте токены в журналы (логи). Партнёрский токен — на сервере, пользовательский — только в памяти запроса.

9. К кому обращаться по вопросам интеграции

По всем вопросам интеграции вы можете обращаться к менеджеру Альбато.