Agent Client Protocol (ACP): обзор для разработчиков

Agent Client Protocol (ACP) — это стандартизированный протокол обмена сообщениями для взаимодействия с искусственными интеллектами-агентами. ACP определяет единый интерфейс коммуникации между кодовыми редакторами (IDE) и AI-агентами, обеспечивая совместимость между различными реализациями без привязки к конкретному поставщику.

Протокол был разработан для решения проблемы фрагментации экосистемы AI-агентов. Ранее каждый редактор кода (Zed, JetBrains IDE, VS Code) требовал собственных интеграций с каждым агентом (Claude Code, Gemini CLI, GitHub Copilot), что приводило к:

• Высоким затратам на интеграцию: Каждая новая комбинация «агент-редактор» требовала написания кастомного кода
• Ограниченной совместимости: Агенты работали только с подмножеством доступных редакторов
• Vendor lock-in: Выбор агента часто означал принятие его интерфейсов без альтернатив

ACP решает эти проблемы, предоставляя открытый стандарт, который позволяет любому агенту, реализующему протокол, беспрепятственно интегрироваться с любым совместимым редактором.

ACP был инициирован командой Zed Industries и получил активную поддержку JetBrains. В октябре 2025 года JetBrains и Zed объединились для публичного объявления о совместной разработке протокола как открытого стандарта для интеграции AI-агентов в IDE.

Ключевые вклады:

• Zed Industries: Изначальная разработка и публикация спецификации протокола в августе 2025 года
• JetBrains: Интеграция ACP в свои IDE (IntelliJ IDEA, PyCharm и другие) и поддержка стандарта в экосистеме

Обе компании подчеркивают открытость стандарта: ACP распространяется под лицензией Apache и доступен для использования любыми разработчиками и организациями.

Что такое Agent Client Protocol?

Agent Client Protocol (ACP) — это JSON-RPC 2.0-совместимый протокол, предназначенный для обмена сообщениями между AI-агентами и клиентскими приложениями (в первую очередь, кодовыми редакторами).

Основные принципы

Архитектурная модель

ACP реализует классическую клиент-серверную модель:

• ACP Client (Клиент): Приложение, инициирующее запросы (обычно кодовый редактор)
• ACP Server (Сервер): AI-агент, обрабатывающий запросы и выполняющий инструменты

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

Поддерживаемые сценарии

• Локальные агенты: Запускаются как подпроцессы редактора, общаются через JSON-RPC по stdio
• Удалённые агенты: Хостятся в облаке или на отдельной инфраструктуре, общаются через HTTP/WebSocket

Примечание: Полная поддержка удалённых агентов находится в разработке. Команда ACP активно сотрудничает с платформами агентов для обеспечения соответствия протокола требованиям облачных развертываний.

История и мотивация

Историческая хронология:

• Август 2025: Zed Industries публично анонсирует ACP и выпускает первую версию спецификации с интеграцией Google Gemini CLI
• Сентябрь 2025: ACP становится доступен в Zed для пользователей
• Октябрь 2025: JetBrains и Zed объявляют о совместной работе над стандартом и планируемой интеграцией в JetBrains IDE
• Ноябрь 2025: Появляются официальные SDK для Python, Rust и TypeScript

До появления ACP экосистема AI-агентов страдала от следующих проблем:

1. Фрагментация API: Каждый агент (Claude Code, GitHub Copilot, Gemini CLI) имел собственный API
2. Отсутствие стандарта: Нет единого способа интеграции агентов в редакторы
3. Повторная реализация: Каждый редактор должен был писать интеграции для каждого агента

Пример до-ACP интеграции

До ACP интеграция Claude Code в Zed требовала:

// Zed должен был иметь специальную интеграцию для Claude Code
{
  "agent_type": "claude_code",
  "api_endpoint": "https://api.anthropic.com/v1/messages",
  "auth_method": "Bearer token",
  "message_format": "claude-specific-schema"
}

А для Gemini CLI — другая интеграция:

// Zed должен был иметь специальную интеграцию для Gemini CLI
{
  "agent_type": "gemini_cli",
  "command": "gemini",
  "args": ["--mode", "chat"],
  "message_format": "gemini-specific-schema"
}

ACP решает эту проблему, предоставляя единый формат сообщений:

// Единый формат для всех агентов через ACP
{
  "jsonrpc": "2.0",
  "method": "session/prompt",
  "params": {
    "session_id": "abc123",
    "content": "Как работает ACP?"
  }
}

Сравнение с альтернативами

ACP vs OpenAI Functions API

Преимущества ACP:

• Независимость от конкретного поставщика LLM
• Поддержка потоковой передачи (streaming) в реальном времени
• Встроенная система управления сессиями и инструментами
• Поддержка локальных и удалённых агентов

ACP vs LangChain Protocols

ACP и LangChain не конкурируют, а дополняют друг друга:

• LangChain может использоваться для реализации логики агента
• ACP используется для коммуникации этого агента с внешним миром (IDE)

ACP vs Custom HTTP APIs

Основные преимущества ACP

1. Открытый стандарт: Apache License, исходный код доступен на GitHub
2. Независимость от поставщика: Работает с любыми LLM (OpenAI, Anthropic, Google, локальные модели)
3. Богатые возможности: Поддержка потоковой передачи, инструментов, сессий, разрешений
4. Кроссплатформенность: SDK для Python, Rust, TypeScript, Go
5. Безопасность: Встроенная система разрешений и аутентификации

Архитектурный обзор

Поток сообщений

ACP поддерживает три основных паттерна коммуникации:

Типы соединений

ACP поддерживает три основных канала коммуникации:

• stdio (stdin/stdout) — основной метод для локальных агентов, запущенных как подпроцессы
• WebSocket — предпочтительный метод для удалённых агентов
• HTTP — используется для запросов-ответов без постоянного соединения

ACP использует JSON-RPC 2.0 как основу для структуры сообщений. Этот протокол обеспечивает простую, но мощную основу для удалённого вызова процедур (RPC), поддерживающую как синхронные запросы-ответы, так и асинхронные уведомления.

Каждое сообщение ACP — это валидный JSON-объект со следующими обязательными полями:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "session/prompt",
  "params": { ... }
}

Поля ответа

Ответ от сервера содержит либо поле result, либо поле error:

// Успешный ответ
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": { ... }
}

// Ответ с ошибкой
{
  "jsonrpc": "2.0",
  "id": 1,
  "error": { ... }
}

Коды ошибок и обработка

ACP поддерживает все стандартные коды JSON-RPC 2.0 и добавляет специфичные для AI-агентов коды.

Стандартные коды JSON-RPC 2.0:

Специфичные коды ACP:

ACP расширяет базовый JSON-RPC 2.0, добавляя поля и методы, специфичные для AI-агентов.

Специфичные методы ACP

ACP определяет следующие основные методы:

Полный список методов вы можете посмотреть на сайте https://agentclientprotocol.com/protocol/schema

Модели данных

Базовые типы данных

ContentBlock

Основной блок контента для сообщений. Является основным контейнером для всех данных, возвращаемых агентом.

SessionUpdate

Обновление состояния сессии. Отправляется сервером клиенту как уведомление.

SessionNotification

Уведомление о событии сессии. Стриминг частей ответа сервера клиенту.

AgentNotification

Уведомление от агента.

Сообщения запросов

session/prompt

Выполняет запрос к агенту и возвращает полный результат в одном ответе.

Руководство по реализации

ACP-клиент — это приложение, которое инициирует запросы к AI-агенту. Обычно это кодовый редактор (Zed, JetBrains IDE, VS Code) или другая интегрированная среда разработки.

Настройка соединения

ACP поддерживает три основных канала коммуникации: stdio, WebSocket и HTTP.

stdio соединение — основной метод для локальных агентов, запущенных как подпроцессы.

WebSocket соединение — предпочтительный метод для удалённых агентов, обеспечивающий двунаправленную коммуникацию в реальном времени.

HTTP соединение — используется для запросов-ответов без постоянного соединения. Подходит для простых сценариев или когда WebSocket недоступен.

Создание ACP-сервера

ACP-сервер — это AI-агент, который получает запросы от клиента, обрабатывает их и выполняет инструменты.

1. Транспортный слой: Обработка соединений (stdio, WebSocket, HTTP)
2. Маршрутизатор методов: Диспетчеризация вызовов методов
3. Агент: Логика обработки запросов и выполнения инструментов
4. Управление сессиями: Хранение и обновление состояния сессий

ACP-сервер может быть интегрирован с различными AI-моделями:

• Облачные модели: OpenAI GPT, Anthropic Claude, Google Gemini
• Локальные модели: Llama, Mistral через Ollama
• Кастомные модели: Собственные модели с API-интерфейсом

ACP-клиент — это приложение, которое инициирует запросы к AI-агенту. Обычно это кодовый редактор (Zed, JetBrains IDE, VS Code), но мы создадим универсальный клиент для демонстрации.

Установка зависимостей

Для работы примеров потребуется библиотека websockets для WebSocket-соединений:

pip install websockets

Код ACP-клиента

"""
ACP Client Example
Полный пример ACP-клиента с поддержкой WebSocket, потоковой передачи
и управления сессиями.
"""

import asyncio
import json
import websockets

from typing import Dict, Any, Optional, List, Callable
from dataclasses import dataclass, field
from enum import Enum
import uuid

class MessageRole(Enum):
    """Роли сообщений"""
    USER = "user"
    ASSISTANT = "assistant"

@dataclass
class ContentBlock:
    """Блок контента сообщения"""
    type: str  # "text", "image", "tool_call", "tool_result"
    text: Optional[str] = None
    tool_call_id: Optional[str] = None
    name: Optional[str] = None
    arguments: Optional[Dict[str, Any]] = None
    is_error: bool = False
    error_message: Optional[str] = None

@dataclass
class Session:
    """Сессия взаимодействия с агентом"""
    session_id: str
    messages: List[Dict[str, Any]] = field(default_factory=list)
    tools: List[Dict[str, Any]] = field(default_factory=list)
    capabilities: Dict[str, Any] = field(default_factory=dict)

class ACPClient:
    """
    ACP-клиент для взаимодействия с AI-агентами.
    Поддерживает WebSocket соединение, потоковую передачу и управление сессиями.
    """

    def __init__(self, websocket_url: str = "ws://localhost:8080/acp"):
        """
        Инициализация клиента.

        Args:
            websocket_url: URL WebSocket-сервера ACP
        """
        self.websocket_url = websocket_url
        self.websocket = None
        self.request_id = 0
        self.pending_requests: Dict[int, asyncio.Future] = {}
        self.event_handlers: Dict[str, Callable] = {}
        self.sessions: Dict[str, Session] = {}
        self.current_session: Optional[Session] = None
        self._receive_task: Optional[asyncio.Task] = None
        self._connected = False

    async def connect(self) -> bool:
        """
        Установка соединения с ACP-сервером.

        Returns:
            True при успешном подключении, False в противном случае
        """
        try:
            self.websocket = await websockets.connect(
                self.websocket_url,
                ping_interval=20,
                ping_timeout=10
            )
            self._connected = True
            # Запуск цикла приёма сообщений
            self._receive_task = asyncio.create_task(self._receive_loop())
            return True
        except Exception as e:
            print(f"Ошибка подключения: {e}")
            return False

    async def disconnect(self):
        """Закрытие соединения с сервером"""
        if self._receive_task:
            self._receive_task.cancel()
            try:
                await self._receive_task
            except asyncio.CancelledError:
                pass

        if self.websocket:
            await self.websocket.close()
            self.websocket = None

        self._connected = False

ACP-сервер — это AI-агент, который получает запросы от клиента, обрабатывает их и выполняет инструменты.

Основные компоненты сервера:

1. Транспортный слой: Обработка соединений (WebSocket, HTTP, IPC)
2. Маршрутизатор методов: Диспетчеризация вызовов методов
3. Агент: Логика обработки запросов и выполнения инструментов
4. Управление сессиями: Хранение и обновление состояния сессий

JSON-диаграммы взаимодействия

Диаграмма: Создание сессии

// Запрос от клиента
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "session/new",
  "params": {
    "capabilities": {
      "tools": {"tool_calling": true},
      "text": {"input": true, "output": true}
    }
  }
}

// Ответ от сервера
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "session_id": "sess_abc123",
    "capabilities": {
      "tools": {"tool_calling": true},
      "text": {"input": true, "output": true}
    },
    "session_state": "active"
  }
}

Диаграмма: Выполнение запроса

// Запрос от клиента
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "session/prompt",
  "params": {
    "session_id": "sess_abc123",
    "content": [
      {
        "type": "text",
        "text": "Какой сегодня день?"
      }
    ]
  }
}

// Ответ от сервера
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "session_id": "sess_abc123",
    "content": [
      {
        "type": "text",
        "text": "Сегодня среда, 9 апреля 2025 года."
      }
    ],
    "stop_reason": "end_turn"
  }
}

Модель процессов и жизненный цикл подключения

Критически важный аспект ACP, который часто упускают в первых реализациях: протокол предполагает долгоживущий (long-running) процесс сервера, а не запуск нового процесса на каждый запрос.

Когда IDE (клиент) подключается к агенту, она запускает один дочерний процесс через stdio (pipes) и поддерживает это соединение на протяжении всей сессии работы с агентом. Это архитектурное решение обусловлено несколькими факторами:

Стандартный поток взаимодействия

┌─────────────────────────────────────────┐
│ 1. IDE запускает процесс агента         │
│    (через ProcessBuilder / subprocess)  │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│ 2. Рукопожатие: initialize              │
│    • Обмен capabilities                 │
│    • Согласование версии протокола      │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│ 3. Создание сессии: session/new         │
│    • Возвращает session_id              │
│    • Инициализирует контекст диалога    │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│ 4. Цикл работы с сессией:               │
│    • session/prompt → ответ или стрим   │
│    • session/notification ← события     │
│    • session/cancel ← отмена (опц.)     │
│    • tools/call → выполнение инструмента│
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│ 5. Завершение: session/destroy          │
│    или закрытие stdin со стороны IDE    │
│    → агент корректно завершает работу   │
└─────────────────────────────────────────┘

Технические требования к транспорту stdio

При реализации сервера над stdin/stdout необходимо соблюдать три строгих правила:

1. Newline-delimited JSON: Каждое сообщение (запрос, ответ, уведомление) должно заканчиваться символом \n. Клиент читает stdout построчно (readline()), и отсутствие \n приведёт к зависанию парсера.

2. Чистый stdout: В поток stdout допускается записывать только валидные JSON-RPC сообщения. Любой print(), traceback, лог отладки или бинарный вывод сломает протокол. Для диагностики используйте stderr.

3. Асинхронная обработка: Чтение stdin и запись в stdout не должны блокировать друг друга. Обработка session/prompt должна запускаться в фоновой задаче (asyncio.create_task), чтобы сервер продолжал принимать новые сообщения (например, session/cancel) во время генерации ответа.

Обработка завершения

Когда пользователь закрывает чат или отключает агента, IDE закрывает stdin процесса. Сервер должен корректно обработать EOF в sys.stdin.readline() как сигнал к завершению:

• Остановить активные задачи генерации
• Сохранить состояние сессий (если требуется)
• Освободить ресурсы (модели, подключения)
• Завершить процесс с кодом 0

Эта модель обеспечивает предсказуемое управление ресурсами и совместимость со всеми клиентами, реализующими спецификацию ACP.

ACP Server (Agent Client Protocol) - Пример демо реализации на python для подключения к IDE (JetBrains, Zed и др.)

"""
ACP Server (Agent Client Protocol)
Пример демо реализации для подключения к IDE (JetBrains, Zed и др.)
Транспорт: JSON-RPC 2.0 over stdio
"""

import asyncio
import json
import sys
import logging
from typing import Dict, Any, Optional
import uuid

# ⚠️ ВАЖНО: Все логи должны идти в stderr. stdout зарезервирован только для JSON-RPC!
logging.basicConfig(
    stream=sys.stderr,
    level=logging.INFO,
    format='%(asctime)s [ACP] %(levelname)s: %(message)s'
)
logger = logging.getLogger(__name__)

class ACPServer:
    def __init__(self, agent_name: str = "CustomAgent", agent_version: str = "1.0.0"):
        self.agent_name = agent_name
        self.agent_version = agent_version
        self.sessions: Dict[str, Dict[str, Any]] = {}
        self._running = True

    async def run(self):
        """Основной цикл чтения сообщений из stdin"""
        logger.info("🟢 ACP Server запущен. Ожидание подключений от IDE...")

        while self._running:
            try:
                # Читаем строку из stdin асинхронно (не блокируя event loop)
                line = await asyncio.to_thread(sys.stdin.readline)
            except EOFError:
                break

            if not line:
                break

            line = line.strip()
            if not line:
                continue

            try:
                message = json.loads(line)
                await self._route_message(message)
            except json.JSONDecodeError as e:
                logger.error(f"❌ Invalid JSON: {e}")
            except Exception as e:
                logger.exception(f"❌ Routing error: {e}")

        logger.info("🔴 ACP Server остановлен.")

    async def _route_message(self, msg: Dict[str, Any]):
        """Маршрутизация входящего JSON-RPC сообщения"""
        if "id" in msg and "method" in msg:
            await self._handle_request(msg)
        elif "method" in msg and "id" not in msg:
            logger.debug(f"Получено уведомление от клиента: {msg.get('method')}")
        else:
            logger.warning(f"Неизвестный формат сообщения: {msg}")

    async def _handle_request(self, msg: Dict[str, Any]):
        """Обработка запросов от IDE"""
        method = msg["method"]
        msg_id = msg["id"]
        params = msg.get("params", {})

        try:
            if method == "initialize":
                result = await self._initialize(params)
            elif method == "session/new":
                result = await self._session_new(params)
            elif method == "session/prompt":
                result = await self._session_prompt(msg_id, params)
            elif method == "session/destroy":
                result = await self._session_destroy(params)
            else:
                await self._send_error(msg_id, -32601, f"Method not supported: {method}")
                return

            await self._send_response(msg_id, result)
        except Exception as e:
            logger.exception(f"Ошибка в {method}")
            await self._send_error(msg_id, -32603, f"Internal error: {str(e)}")

    # ─────────────── ACP МЕТОДЫ ───────────────

    async def _initialize(self, params: Dict) -> Dict:
        """Рукопожатие: обмен capabilities"""
        logger.info("🤝 Initialize запрос от IDE")
        return {
            "protocolVersion": "1.0",
            "agentName": self.agent_name,
            "agentVersion": self.agent_version,
            "capabilities": {
                "streaming": True,
                "tools": ["file_read", "file_write", "run_terminal"],
                "context_awareness": True
            }
        }

    async def _session_new(self, params: Dict) -> Dict:
        """Создание новой сессии диалога"""
        session_id = str(uuid.uuid4())
        self.sessions[session_id] = {
            "id": session_id,
            "history": [],
            "context_files": []
        }
        logger.info(f"📝 Создана сессия: {session_id}")
        return {"sessionId": session_id}

    async def _session_prompt(self, request_id: int, params: Dict) -> Dict:
        """Получение промпта. Запускает генерацию в фоне."""
        session_id = params.get("sessionId")
        prompt = params.get("prompt", [])

        if not session_id or session_id not in self.sessions:
            await self._send_error(request_id, -32602, "Invalid sessionId")
            return None

        # Сохраняем пользовательский запрос в историю
        self.sessions[session_id]["history"].append({"role": "user", "content": prompt})

        # Запускаем генерацию асинхронно, чтобы не блокировать stdin
        asyncio.create_task(self._generate_and_stream(session_id, prompt))

        # ACP требует немедленного подтверждения для streaming-запросов
        return {"sessionId": session_id, "status": "accepted"}

    async def _session_destroy(self, params: Dict) -> Dict:
        """Завершение сессии"""
        session_id = params.get("sessionId")
        self.sessions.pop(session_id, None)
        logger.info(f"🗑️ Сессия закрыта: {session_id}")
        return {"success": True}

    # ─────────────── ЛОГИКА АГЕНТА ───────────────

    async def _generate_and_stream(self, session_id: str, prompt: list):
        """
        ЗДЕСЬ ПОДКЛЮЧАЕТСЯ ТВОЙ РЕАЛЬНЫЙ LLM/АГЕНТ.
        Пока используем эмуляцию стриминга для демонстрации.
        """
        logger.info(f"🧠 Начинаю генерацию для сессии {session_id}")

        # Пример: симуляция токенов от модели
        chunks = ["Привет! ", "Я ", "твой ", "кастомный ", "ACP-агент. ", "Готов ", "помочь. 🚀"]

        for chunk in chunks:
            await asyncio.sleep(0.4)  # Имитация задержки модели
            await self._send_chunk(session_id, chunk)

        # Сигнал окончания стрима
        await self._send_chunk(session_id, "", done=True)
        logger.info("✅ Генерация завершена.")

    async def _send_chunk(self, session_id: str, text: str, done: bool = False):
        """Отправка чанка ответа в IDE через notification"""
        update = {
            "sessionUpdate": "agentMessageChunk",
            "chunk": {
                "content": {"type": "text", "text": text},
                "role": "assistant",
                "timestamp": asyncio.get_event_loop().time()
            }
        }
        if done:
            update["chunk"]["done"] = True

        await self._send_notification(session_id, update)

    # ─────────────── СЕРВИСНЫЕ МЕТОДЫ ───────────────

    async def _send_response(self, msg_id: Any, result: Dict):
        await self._write_json({"jsonrpc": "2.0", "id": msg_id, "result": result})

    async def _send_error(self, msg_id: Any, code: int, message: str):
        await self._write_json({
            "jsonrpc": "2.0",
            "id": msg_id,
            "error": {"code": code, "message": message}
        })

    async def _send_notification(self, session_id: str, params: Dict):
        await self._write_json({
            "jsonrpc": "2.0",
            "method": "session/notification",
            "params": {"sessionId": session_id, "update": params}
        })

    async def _write_json(self, obj: Dict):
        """Атомарная запись JSON в stdout с переводом строки"""
        loop = asyncio.get_running_loop()
        data = json.dumps(obj, ensure_ascii=False) + "\n"
        await asyncio.to_thread(sys.stdout.write, data)
        await asyncio.to_thread(sys.stdout.flush)

if __name__ == "__main__":
    # Запуск сервера. Python 3.9+
    server = ACPServer(agent_name="MyPythonAgent", agent_version="0.1.0")
    try:
        asyncio.run(server.run())
    except KeyboardInterrupt:
        server._running = False
    except Exception as e:
        logger.exception("Критическая ошибка сервера")
        sys.exit(1)

Сценарии использования и приложения

Интеграция ACP в интегрированные среды разработки (IDE) — один из ключевых сценариев использования протокола. ACP обеспечивает единый интерфейс между редакторами кода и AI-агентами, позволяя разработчикам получать помощь от искусственного интеллекта непосредственно в своей рабочей среде.

Интеграция с JetBrains IDE

Интеграция позволяет использовать любые ACP-совместимые агенты (Claude Code, Gemini CLI, локальные модели) без необходимости установки специальных плагинов для каждого агента.

Интеграция с Zed Editor

Zed Editor был первым редактором, который получил полную поддержку ACP. Интеграция реализована на уровне ядра редактора и позволяет использовать любые ACP-совместимые агенты.

Как ACP включает AI-функции в IDE

ACP позволяет IDE реализовывать следующие AI-функции:

ACP обеспечивает совместимость между различными агентными фреймворками, позволяя разработчикам использовать привычные инструменты для построения агентов, а затем интегрировать их в IDE через стандартный протокол.

ACP находит применение в корпоративных решениях:

• Кастомные агенты для внутренних нужд
• Интеграция с существующими системами
• Управление доступом и аудит

Ссылки и приложения

1. Zed Industries. Agent Client Protocol Specification. Август 2025.
2. Agent Client Protocol Official Website.
3. JetBrains. JetBrains AI Assistant — ACP Documentation.
4. PyPI. agent-client-protocol Python SDK.
5. GitHub. agentclientprotocol/agent-client-protocol Repository
6. JSON-RPC 2.0 Specification.
7. OpenAI. Chat Completions API.
8. WebSocket Protocol (RFC 6455).

Определения типов сообщений

В этом приложении представлены JSON-схемы типов сообщений, используемых в Agent Client Protocol (ACP). Все схемы соответствуют стандарту JSON-RPC 2.0 с расширениями ACP.

Базовая структура сообщения

Каждое сообщение ACP — это валидный JSON-объект со следующей базовой структурой:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "session/prompt",
  "params": { ... }
}

Поля:

• jsonrpc (string, обязательное): Версия протокола JSON-RPC. Всегда "2.0" для ACP.
• id (integer/string, обязательное): Уникальный идентификатор запроса. Должен быть уникален в пределах одной сессии.
• method (string, обязательное): Имя вызываемого метода. Методы имеют префикс session/, fs/, terminal/ или event/.
• params (object/array/null, опциональное): Параметры метода.

Запросы (Requests)

session/prompt

Выполняет агента и возвращает полный результат в одном ответе.

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "session/prompt",
  "params": {
    "session_id": "sess_abc123",
    "content": [
      {
        "type": "text",
        "text": "Какой сегодня день?"
      }
    ],
    "metadata": {
      "priority": "normal",
      "timeout_ms": 30000
    }
  }
}

Параметры:

• session_id (string, обязательное): ID сессии для контекста.
• content (array, обязательное): Массив блоков контента запроса.
• metadata (object, опциональное): Метаданные запроса.
• timeout_ms (integer, опциональное): Таймаут в миллисекундах.

session/new

Создает новую сессию.

{
  "jsonrpc": "2.0",
  "id": 5,
  "method": "session/new",
  "params": {
    "capabilities": {
      "tools": {
        "tool_calling": true
      },
      "text": {
        "input": true,
        "output": true
      }
    }
  }
}

Параметры:

• capabilities (object, опциональное): Поддерживаемые возможности клиента.

Ответы (Responses)

session/prompt_response

Ответ на запрос session/prompt.

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "session_id": "sess_abc123",
    "content": [
      {
        "type": "text",
        "text": "Сегодня среда, 9 апреля 2025 года."
      }
    ],
    "stop_reason": "end_turn",
    "metadata": {
      "tokens_used": 256,
      "execution_time_ms": 1523
    }
  }
}

Поля результата:

• session_id (string, обязательное): ID сессии.
• content (array, обязательное): Массив блоков результата.
• stop_reason (string, обязательное): Причина остановки ("end_turn", "max_tokens", "tool_calls").
• metadata (object, опциональное): Метаданные ответа.

Уведомления (Notifications)

session/update

Обновление состояния сессии.

{
  "jsonrpc": "2.0",
  "method": "session/update",
  "params": {
    "session_id": "sess_abc123",
    "update": {
      "type": "state",
      "state": {
        "turn_count": 5,
        "last_message_id": "msg_abc123",
        "tokens_used": 1250
      }
    }
  }
}

Обработка ошибок

Обработка ошибок является критически важной частью реализации ACP. Протокол предоставляет детальную систему кодов ошибок и рекомендации по их обработке как на стороне клиента, так и на стороне сервера.

Классификация ошибок

JSON-RPC 2.0 стандартные ошибки

ACP-специфичные ошибки

Структура сообщения об ошибке

// JSON-RPC 2.0 структура

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "error": {
    "code": -32601,
    "message": "Method not found",
    "data": {
      "method": "session/prompt",
      "timestamp": "2025-04-09T10:00:00Z",
      "request_id": "request-id"
    }
  }
}

// ACP расширенная структура

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "error": {
    "code": -32002,
    "message": "Session not found",
    "data": {
      "session_id": "session-uuid",
      "timestamp": "2025-04-09T10:00:00Z",
      "request_id": "request-id",
      "suggested_action": "Create new session",
      "retry_after": 5
    }
  }
}

Стратегии обработки ошибок

// Обработка JSON-RPC ошибок

try:
    data = json.loads(message)
except json.JSONDecodeError as e:
    error_response = {
        "jsonrpc": "2.0",
        "id": None,
        "error": {
            "code": -32700,
            "message": "Parse error",
            "data": {
                "error": str(e),
                "received_message": message[:200]
            }
        }
    }
    await websocket.send(json.dumps(error_response))

// Пример обработки ACP ошибок

async def handle_execute(message: Dict[str, Any]) -> Dict[str, Any]:
    params = message.get("params", {})
    session_id = params.get("session_id")

    session = await session_manager.get_session(session_id)
    if not session:
        return {
            "jsonrpc": "2.0",
            "id": message.get("id"),
            "error": {
                "code": -32002,
                "message": "Session not found",
                "data": {
                    "session_id": session_id,
                    "suggested_action": "Call session/new first"
                }
            }
        }
    # Обработка запроса...