OpenAI-совместимый API: проверка JSON, tools и streaming

OpenAI-совместимый API — это конкретный контракт запроса и ответа. Через шлюз может работать короткий текст, но не работать JSON Schema или возврат результата инструмента. Поэтому перед переносом приложения нужно проверить именно те функции, которыми оно пользуется.

Ниже — матрица Chat Completions, Responses и Anthropic Messages, один Python-файл с четырьмя отдельными проверками и условия успешного результата. Это документальная инструкция для самостоятельного запуска. Редакция проверила чтение каталога Zenith и локальную работу обработчиков на искусственных ответах; платные текстовые запросы по этим сценариям не выполняла.

Как отличаются три протокола

Пути, базовые адреса и заголовки авторизации опубликованы в справочнике Zenith. Названия полей взяты из руководства OpenAI по миграции, описания function calling, Structured outputs Anthropic и цикла tool_use/tool_result. Это исходные контракты производителей, а не доказательство всех функций шлюза.

Для SDK Anthropic к Base URL не добавляют /v1: клиент формирует путь сам. В примере ниже используется raw HTTP, поэтому полный путь /v1/messages указан явно. Точно так же SDK-свойство output_text в Responses нельзя искать как обязательное поле сырого JSON: здесь собираются текстовые блоки массива output.

Как выбрать модель, не угадывая поддержку

Сначала получите ключ через бота Zenith и уточните условия предложенной операции до оплаты. Эта статья начинается с уже выданного ключа и доступного баланса. Скрипт читает GET /v1/models и останавливается, если выбранного ID или протокола нет в supported_endpoint_types. В редакционном чтении 28 октября 2026 года каталог вернул HTTP200, объект с массивом data; результат относится к одному ключу и моменту времени. Это единичное наблюдение wire-формата, не гарантированный контракт или будущая схема. Неизвестный корневой формат останавливает скрипт до POST.

В историческом снимке 27 октября gpt-6-luna указан с chat_completions и responses, claude-haiku-4-5 — с messages. Снимок относится к одному ключу и моменту времени, не устанавливает доступность для другого ключа. Это кандидаты для проверки, а не обещание сегодняшней доступности или установленного backend. Метаданные каталога не подтверждают JSON Schema, инструменты и поток. Каждый такой режим нужно проверять отдельно.

Есть и ограничения исходного API: по документации OpenAI GPT-6 Astra и GPT-6.1 Sol требуют Responses для вызова инструментов; примеры Chat с GPT-5.6 Terra используют reasoning=none. Поэтому проба Chat-tools для GPT ниже требует none в каталоге и явно передаёт reasoning_effort=none. Если режима нет, выберите Responses, не пытайтесь механически повторять тот же запрос.

Один файл: четыре независимые проверки

Нужен Python 3 и requests: установите библиотеку командой python -m pip install requests. Сохраните код как protocol_probe.py. Каждый запуск читает каталог и делает один платный POST; режим tools — максимум два POST, потому что результат функции нужно вернуть модели. Автоматических повторов и перенаправления ключа на другой адрес нет. Лимит вывода 1024 токена может оказаться мал для отдельной reasoning-модели: незавершённый ответ считается неудачей этой проверки.

"""Documentary Zenith probes. Importing this file never performs HTTP."""
import copy
import getpass
import json
import sys

BASE = "https://api.zenithmarket.dev"
PATHS = {"chat_completions": "/v1/chat/completions", "responses": "/v1/responses", "messages": "/v1/messages"}

def make_payload(protocol, mode, model):
    prompt = {"text": "Reply exactly: ok", "stream": "Reply exactly: ok",
              "json": "Return only JSON with result equal to 4.",
              "tools": "Call add with a=2,b=2. After its result reply exactly: 4."}[mode]
    user = {"role": "user", "content": prompt}
    body = {"model": model, "input": prompt, "max_output_tokens": 1024} if protocol == "responses" else {
        "model": model, "messages": [user], "max_tokens": 1024}
    schema = {"type": "object", "properties": {"result": {"type": "integer"}},
              "required": ["result"], "additionalProperties": False}
    if mode == "json":
        if protocol == "chat_completions":
            body["response_format"] = {"type": "json_schema", "json_schema": {
                "name": "answer", "strict": True, "schema": schema}}
        elif protocol == "responses":
            body["text"] = {"format": {"type": "json_schema", "name": "answer", "strict": True, "schema": schema}}
        else:
            body["output_config"] = {"format": {"type": "json_schema", "schema": schema}}
    if mode == "tools":
        parameters = {"type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
                      "required": ["a", "b"], "additionalProperties": False}
        fn = {"name": "add", "description": "Add two integers; returns their sum.", "parameters": parameters, "strict": True}
        if protocol == "chat_completions":
            body["tools"] = [{"type": "function", "function": fn}]
        elif protocol == "responses":
            body["tools"] = [{"type": "function", **fn}]
        else:
            body["tools"] = [{"name": "add", "description": fn["description"], "input_schema": parameters, "strict": True}]
    if mode == "stream": body["stream"] = True
    return body

def text_of(protocol, result):
    if protocol == "chat_completions":
        if result["choices"][0].get("finish_reason") != "stop": raise ValueError("Необычное завершение Chat")
        return result["choices"][0]["message"].get("content") or ""
    if protocol == "responses":
        if result.get("status") != "completed": raise ValueError("Responses не завершён")
        return "".join(b.get("text", "") for item in result.get("output", []) if item.get("type") == "message"
                       for b in item.get("content", []) if b.get("type") == "output_text")
    if result.get("stop_reason") != "end_turn": raise ValueError("Messages не завершён")
    return "".join(b.get("text", "") for b in result.get("content", []) if b.get("type") == "text")

def tool_continuation(protocol, original, result):
    body = copy.deepcopy(original)
    if protocol == "chat_completions":
        if result["choices"][0].get("finish_reason") != "tool_calls": raise ValueError("Незавершённый вызов инструмента Chat")
        message = result["choices"][0]["message"]; calls = message.get("tool_calls", [])
        if len(calls) != 1: raise ValueError("Нужен один tool call")
        call = calls[0]; name = call["function"]["name"]; args = json.loads(call["function"]["arguments"]); ident = call["id"]
    elif protocol == "responses":
        if result.get("status") != "completed": raise ValueError("Незавершённый Responses")
        calls = [x for x in result.get("output", []) if x.get("type") == "function_call"]
        if len(calls) != 1: raise ValueError("Нужен один function_call")
        call = calls[0]; name = call["name"]; args = json.loads(call["arguments"]); ident = call["call_id"]
    else:
        if result.get("stop_reason") != "tool_use": raise ValueError("Незавершённый tool_use")
        calls = [x for x in result.get("content", []) if x.get("type") == "tool_use"]
        if len(calls) != 1: raise ValueError("Нужен один tool_use")
        call = calls[0]; name = call["name"]; args = call["input"]; ident = call["id"]
    if name != "add" or not isinstance(args, dict) or set(args) != {"a", "b"}:
        raise ValueError("Неизвестный инструмент или схема")
    if any(type(args[k]) is not int or args[k] != 2 for k in ("a", "b")) or not isinstance(ident, str) or not ident:
        raise ValueError("Неправильные аргументы/ID")
    output = str(args["a"] + args["b"])  # only this deterministic allowlisted operation
    if protocol == "chat_completions":
        body["messages"] += [{"role": "assistant", "content": message.get("content"), "tool_calls": calls},
                             {"role": "tool", "tool_call_id": ident, "content": output}]
    elif protocol == "responses":
        body["input"] = [{"role": "user", "content": original["input"]}] + result["output"] + [
            {"type": "function_call_output", "call_id": ident, "output": output}]
    else:
        body["messages"] += [{"role": "assistant", "content": result["content"]},
                             {"role": "user", "content": [{"type": "tool_result", "tool_use_id": ident, "content": output}]}]
    return body

def sse_events(response):
    event, data = None, []
    for raw in response.iter_lines():
        line = raw.decode("utf-8")
        if not line:
            if data: yield event, "\n".join(data)
            event, data = None, []
        elif line.startswith("event:"): event = line[6:].lstrip()
        elif line.startswith("data:"): data.append(line[5:].lstrip())

def stream_text(protocol, response):
    parts, completed, reason = [], False, None
    for event, raw in sse_events(response):
        if raw == "[DONE]": continue
        obj = json.loads(raw); typ = obj.get("type", event)
        if obj.get("error") or typ in {"error", "response.failed", "response.incomplete"}: raise ValueError("Ошибка в потоке")
        if protocol == "chat_completions":
            for choice in obj.get("choices", []):
                parts.append(choice.get("delta", {}).get("content") or "")
                if choice.get("finish_reason") is not None: completed = choice["finish_reason"] == "stop"
        elif protocol == "responses":
            if typ == "response.output_text.delta": parts.append(obj.get("delta", ""))
            if typ == "response.completed": completed = True
        else:
            delta = obj.get("delta", {})
            if typ == "content_block_delta" and delta.get("type") == "text_delta": parts.append(delta.get("text", ""))
            if typ == "message_delta": reason = delta.get("stop_reason", reason)
            if typ == "message_stop": completed = reason == "end_turn"
    if not completed: raise ValueError("Поток не получил нормального завершения")
    return "".join(parts)

def main():
    import requests
    if len(sys.argv) != 4 or sys.argv[1] not in PATHS or sys.argv[2] not in {"text", "json", "tools", "stream"}:
        raise SystemExit("python protocol_probe.py chat_completions|responses|messages text|json|tools|stream MODEL_ID")
    protocol, mode, model = sys.argv[1:]; key = getpass.getpass("Zenith API key: ")
    if not key: raise SystemExit("Ключ не введён")
    headers = {"Authorization": "Bearer " + key, "User-Agent": "Comrad404SourceReview/1.0", "Content-Type": "application/json"}
    if protocol == "messages": headers["anthropic-version"] = "2023-06-01"
    with requests.get(BASE + "/v1/models", headers=headers, timeout=30, allow_redirects=False) as r:
        if r.status_code != 200: raise ValueError("Каталог: HTTP " + str(r.status_code))
        catalog = r.json()
    if isinstance(catalog, list): rows = catalog
    elif isinstance(catalog, dict) and isinstance(catalog.get("data"), list): rows = catalog["data"]
    else: raise ValueError("Неизвестная схема каталога")
    row = next((x for x in rows if x.get("id") == model), None)
    if not row or protocol not in row.get("supported_endpoint_types", []): raise ValueError("Нет ID/endpoint для ключа")
    body = make_payload(protocol, mode, model)
    if protocol == "chat_completions" and mode == "tools" and model.startswith("gpt-"):
        if "none" not in row.get("reasoningEffortModes", []): raise ValueError("Для этого GPT Chat-tools нужен none; проверьте Responses")
        body["reasoning_effort"] = "none"
    def send(payload, streaming=False):
        r = requests.post(BASE + PATHS[protocol], json=payload, headers=headers, timeout=(30, 180), stream=streaming, allow_redirects=False)
        print("HTTP", r.status_code, "requestId", r.headers.get("x-request-id"))
        if r.status_code != 200:
            text = r.text.replace(key, "[REDACTED]")[:800]; r.close(); raise ValueError(text)
        return r
    with send(body, mode == "stream") as first:
        if mode == "stream":
            text = stream_text(protocol, first)
        else: result = first.json()
    if mode == "tools":
        with send(tool_continuation(protocol, body, result)) as second: result = second.json()
        text = text_of(protocol, result)
    elif mode != "stream": text = text_of(protocol, result)
    if mode == "json":
        obj = json.loads(text)
        if obj != {"result": 4} or type(obj.get("result")) is not int: raise ValueError("Неверный JSON/эталон")
    elif text.strip() != ("4" if mode == "tools" else "ok"): raise ValueError("Ответ не выполнил задание")
    print(mode + "_contract=OK")

if __name__ == "__main__":
    try: main()
    except Exception as error:
        print(type(error).__name__, str(error)); raise SystemExit("Проверка не пройдена; автоматического повтора нет")

Запуск: python protocol_probe.py responses text MODEL_ID. Замените MODEL_ID на точный ID из своего каталога. Затем отдельно повторите команду с json, tools или stream. Для других протоколов первый аргумент — chat_completions либо messages. Ключ вводится скрыто; не вставляйте его в команду и не присылайте журналы с секретом.

Что считать успешным результатом

HTTP200 сам по себе не проходит эти критерии. Скрипт не исполняет произвольную функцию от модели: разрешена только сумма двух заданных целых чисел. Для Responses он возвращает весь output, включая reasoning-элементы, а для Messages — весь assistant content, чтобы не потерять необходимые блоки истории. Синтетическая локальная проверка этих обработчиков прошла 32 проверок, включая отказ неизвестному инструменту, неверным аргументам и оборванному потоку. Она не является проверкой удалённого сервиса.

Потоковые события различаются: Chat заканчивает ответ finish_reason=stop, Responses — response.completed, Messages — message_stop после stop_reason=end_turn. Обработчик объединяет многострочные data одного SSE-события. Ошибка внутри потока или обрыв не превращаются в успех из-за ранее полученного HTTP200. Подробности исходного Messages-контракта — в документации Anthropic по streaming.

JSON Schema проверяет структуру, но не истинность значения: здесь число 4 дополнительно сверяется с ручным эталоном. Настоящему приложению понадобятся собственные задачи и проверки. Успех простого add также не доказывает работу параллельных функций, длинного агентного цикла, поиска или инструментов конкретной IDE.

Сколько стоит проверка и как избежать двойного запроса

Для исторических ставок Zenith gpt-6-luna от 27 октября — $0,016 за миллион входных и $0,08 за миллион выходных токенов — условные 4000/800 стоят 4000 × 0,016 / 1000000 + 800 × 0,08 / 1000000 = $0,000128. Десять тысяч таких обращений — $1,28. Два обращения с тем же условным объёмом каждое стоят $0,000256, десять тысяч пар — $2,56. Это арифметическая нагрузка, не измеренные токены короткого задания: второй ход tools включает историю, поэтому его расход обычно отличается.

У claude-haiku-4-5 в том же снимке Zenith ставки $0,34/$1,70: условные 4000/800 дают $0,00272 за одно обращение. Не переносите цену одного ID на другой. Кэш, дополнительные инструменты и повторные запросы здесь не учтены; актуальную стоимость берут из текущего каталога и истории сервиса, а не из этой исторической таблицы.

При HTTP-ошибке прочитайте её тело; при таймауте исход может быть неизвестен. Сохраните время, ID модели и x-request-id, если он возвращён, затем проверьте историю списаний. Отсутствие заголовка или записи не доказывает бесплатность. Не включайте автоматический повтор платного POST до выяснения результата предыдущего.

Как использовать результат в приложении

Запишите дату, клиент, ID, endpoint и результат каждого нужного режима. Для приложения с функциями мало строки text_contract=OK: нужен tools_contract=OK с корректным возвратом результата. Для интерфейса с потоком отдельно нужен stream_contract=OK. Если нужная функция не проходит, оставьте прежнего провайдера или меняйте контракт приложения; один общий ключ не делает три протокола взаимозаменяемыми.