Багатораундові запити

Інколи інструмент не може завершити роботу за один раунд обміну. Йому потрібне щось, що є лише в користувача: вибір, підтвердження, облікові дані.

До версії 2026-07-28 сервер отримував це через зворотний виклик: відкривав власний запит до клієнта — еліцитацію (elicitation) чи виклик семплювання (sampling) — просто посеред обробки початкового запиту. Специфікація 2026-07-28 цей зворотний канал (back-channel) прибирає.

Натомість сервер повертає результат.

Повернення замість зворотного виклику

На tools/call сервер відповідає InputRequiredResult замість CallToolResult. Усю роботу виконують два його поля:

• input_requests: те, чого серверу ще бракує, у вигляді словника з ключами-іменами, які обрав сам сервер. Кожне значення — це ElicitRequest, CreateMessageRequest або ListRootsRequest.
• request_state: непрозорий токен. Під час повторної спроби клієнт повертає його дослівно. Читає його лише ваш сервер.

Клієнт виконує кожен запит, а потім викликає той самий інструмент ще раз, передаючи відповіді в input_responses, а токен — у request_state. Тепер сервер має те, чого йому бракувало, і повертає звичайний CallToolResult.

Оце й увесь протокол. Кожен етап — це звичайний запит від клієнта до сервера. У зворотному напрямку не йде нічого.

Серверна частина

З @mcp.tool() збирати це вручну доводиться рідко: оголосіть залежність — Elicit, щоб запитати користувача, Sample, щоб виконати семплювання через LLM клієнта, або ListRoots, щоб отримати перелік його кореневих каталогів (roots), — і SDK сам поверне InputRequiredResult; цю форму описано на сторінці Залежності. Ці дві форми не поєднуються: у виклику є лише один канал input_responses/request_state, тож інструмент із параметрами Resolve(...) не може ще й повертати InputRequiredResult зі свого тіла. Оголошений тип повернення InputRequiredResult відхиляється під час реєстрації (InvalidSignature), а неоголошений провалює виклик під час виконання. Ручна форма — це низькорівневий Server, чий обробник on_call_tool може повертати будь-який із двох типів результату:

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ElicitRequest,
    ElicitRequestFormParams,
    ElicitResult,
    InputRequiredResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

ASK_REGION = ElicitRequest(
    params=ElicitRequestFormParams(
        message="Which region should the database live in?",
        requested_schema={
            "type": "object",
            "properties": {"region": {"type": "string"}},
            "required": ["region"],
        },
    )
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(
        tools=[
            Tool(
                name="provision",
                description="Provision a database. Asks which region to put it in.",
                input_schema={
                    "type": "object",
                    "properties": {"name": {"type": "string"}},
                    "required": ["name"],
                },
            )
        ]
    )


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult | InputRequiredResult:
    answer = (params.input_responses or {}).get("region")
    if not isinstance(answer, ElicitResult) or answer.content is None:
        return InputRequiredResult(input_requests={"region": ASK_REGION}, request_state="provision-v1")
    name = (params.arguments or {})["name"]
    text = f"Provisioned {name!r} in {answer.content['region']}."
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Provisioner", on_list_tools=list_tools, on_call_tool=call_tool)

• on_call_tool має анотацію -> CallToolResult | InputRequiredResult. Повернути другий із них — оце й увесь серверний API.
• Під час першого виклику params.input_responses дорівнює None, тож спрацьовує перевірка й обробник запитує замість того, щоб відповідати.
• Під час повторної спроби ElicitResult, який надіслав клієнт, лежить під тим самим ключем ("region"), який сервер використав у input_requests.

Усе інше в цьому файлі (явна input_schema, зібраний вручну CallToolResult) — звичайний низькорівневий Server, описаний на сторінці Низькорівневий Server. Ця сторінка додає лише другий тип повернення.

Не лише інструменти

tools/call нічим не особливий: у версії 2026-07-28 сервер може так само відповідати на prompts/get і resources/read. У MCPServer функція @mcp.prompt() — або функція-шаблон @mcp.resource() — сама повертає InputRequiredResult і зчитує відповіді повторної спроби з контексту:

from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.prompts.base import UserMessage
from mcp.types import ElicitRequest, ElicitRequestFormParams, ElicitResult, InputRequiredResult

mcp = MCPServer("Briefing")

ASK_AUDIENCE = ElicitRequest(
    params=ElicitRequestFormParams(
        message="Who is the briefing for?",
        requested_schema={
            "type": "object",
            "properties": {"audience": {"type": "string"}},
            "required": ["audience"],
        },
    )
)


@mcp.prompt()
async def briefing(ctx: Context) -> list[UserMessage] | InputRequiredResult:
    """Draft a briefing tuned to its audience."""
    answer = (ctx.input_responses or {}).get("audience")
    if not isinstance(answer, ElicitResult) or answer.content is None:
        return InputRequiredResult(input_requests={"audience": ASK_AUDIENCE})
    return [UserMessage(f"Write a briefing for {answer.content['audience']}.")]

• Перший раунд повертає InputRequiredResult. Під час повторної спроби ctx.input_responses містить відповіді під тими самими ключами, і функція повертає свій звичайний результат — тут це повідомлення промпту, а для шаблонного ресурсу — вміст ресурсу.
• Заданий вами request_state запечатується, перш ніж потрапити в мережу, і перевіряється, коли повертається, як і все інше на сервері; розділ Захист requestState нижче пояснює, що дає запечатування і коли потрібно налаштовувати ключі.
• Функція @mcp.tool() може так само повертати результат напряму, коли форма із залежностями не підходить.
• Статичні функції @mcp.resource() участі не беруть: вони не приймають Context, тож ніяк не могли б прочитати повторну спробу. Запитувати можуть лише шаблонні ресурси.
• Правила поколінь, наведені нижче, діють без змін: повернення InputRequiredResult у сесії, старшій за 2026, дає ту саму помилку -32603, яку описує попередження.

Клієнтська частина

Client виконує цикл за вас.

Зареєструйте колбеки, які можуть знадобитися серверу (elicitation_callback, sampling_callback, list_roots_callback), і викличте інструмент. Коли надходить InputRequiredResult, Client передає кожен запис із input_requests відповідному колбеку, повторює виклик із відповідями та повернутим request_state і продовжує, доки не надійде CallToolResult:

from mcp import Client
from mcp.client import ClientRequestContext
from mcp.types import ElicitRequestParams, ElicitResult


async def handle_elicitation(context: ClientRequestContext, params: ElicitRequestParams) -> ElicitResult:
    return ElicitResult(action="accept", content={"region": "eu-west-1"})


async def main() -> None:
    async with Client("http://127.0.0.1:8000/mcp", elicitation_callback=handle_elicitation) as client:
        result = await client.call_tool("provision", {"name": "orders"})
        print(result.content)

• Цей elicitation_callback — той самий, у який влучив би elicitation/create зворотного каналу сервера до 2026. Те саме стосується sampling_callback для sampling/createMessage і list_roots_callback для roots/list: у версії 2026-07-28 окремих RPC від сервера до клієнта вже немає, але ідентичні корисні навантаження ElicitRequest / CreateMessageRequest / ListRootsRequest їдуть усередині input_requests і потрапляють до тих самих трьох колбеків. Один набір колбеків обслуговує обидва покоління.
• call_tool повертає звичайний CallToolResult. Проміжні раунди для того, хто викликає, невидимі.
• get_prompt і read_resource запускають той самий цикл.

Цикл обмежений. Client(..., input_required_max_rounds=10) — це ліміт за замовчуванням; якщо сервер і після нього продовжує повертати InputRequiredResult, call_tool викидає виняток. Якщо раунд містить лише request_state без input_requests, Client робить коротку паузу (50 мс, що подвоюється до стелі 250 мс) перед повторною спробою, щоб сервер, який лише каже «ще не готово», не засипали безперервними опитуваннями.

Керування циклом власноруч

Автоматичного циклу достатньо для клієнта в одному процесі. Беріть цикл у свої руки, коли:

• Клієнт розподілений: процес, що показує запитання користувачеві, — не той процес, що викликав call_tool, тож повторну спробу надсилає інший робочий процес. request_state — це придатний до зберігання токен, який ви переносите через цю межу у власному сховищі, а input_responses — те, що інша сторона надсилає разом із ним.
• Потрібно перевіряти кожен раунд: логувати чи аудіювати кожен запис input_requests, відхиляти певні види запитів або застосовувати власну затримку між етапами.
• Потрібне обмеження за реальним часом, а не за кількістю раундів: обгорніть власний цикл у anyio.fail_after(...) замість того, щоб покладатися на input_required_max_rounds.

Спустіться до сесії рівнем нижче, де allow_input_required=True віддає вам об'єднання типів напряму:

from mcp import Client
from mcp.types import CallToolResult, ElicitRequest, ElicitResult, InputRequest, InputRequiredResult, InputResponse


def fulfil(request: InputRequest) -> InputResponse:
    if not isinstance(request, ElicitRequest):
        raise NotImplementedError(f"this client cannot answer a {request.method!r} request")
    return ElicitResult(action="accept", content={"region": "eu-west-1"})


async def provision(client: Client, name: str) -> CallToolResult:
    result = await client.session.call_tool("provision", {"name": name}, allow_input_required=True)
    while isinstance(result, InputRequiredResult):
        responses = {key: fulfil(request) for key, request in (result.input_requests or {}).items()}
        result = await client.session.call_tool(
            "provision",
            {"name": name},
            input_responses=responses,
            request_state=result.request_state,
            allow_input_required=True,
        )
    return result

• client.session.call_tool(..., allow_input_required=True) розширює тип повернення до CallToolResult | InputRequiredResult. Звужує його назад саме isinstance.
• request_state тепер у ваших руках. Збережіть його між етапами — і розмову можна продовжити з нового процесу.
• Для кожного запису в input_requests ви кладете InputResponse під тим самим ключем в input_responses. fulfil — місце для вашого UI; тут відповідь жорстко закодована.
• Та сама назва інструмента, ті самі arguments на кожному етапі. Повторна спроба — це той самий початковий виклик, виконаний ще раз, а не новий метод.

Захист requestState

Усе сказане вище трактує request_state як відлуння, і в переданих даних це справді все, чим він є. Але між етапами його тримає клієнт (а зберігати його між процесами — саме те, що схвалив попередній розділ), тож назад приходять дані, надані клієнтом: їх могли змінити, вони могли прострочитися або взагалі бути взяті з іншого виклику. Специфікація вимагає, щоб сервери захищали цілісність цього стану й відхиляли раунд, якщо перевірка не пройшла, — щоразу, коли стан може впливати на авторизацію, доступ до ресурсів або бізнес-логіку.

MCPServer захищає його за замовчуванням. Кожен сервер запечатує вихідний requestState і перевіряє кожне відлуння — і стан резолверів, і стан, зібраний вручну, — ключем, згенерованим під час запуску процесу. Ви нічого не налаштовуєте, пишете відкритий текст і читаєте відкритий текст; мережею завжди передається лише непрозорий зашифрований токен.

Ключ за замовчуванням живе й помирає разом із процесом — і це єдине, що треба знати перед розгортанням поза межами одного процесу:

from mcp.server.mcpserver import MCPServer, RequestStateSecurity

# Multi-instance or restart-surviving: one or more shared secret keys (>= 32 bytes each).
mcp = MCPServer("fleet", request_state_security=RequestStateSecurity(keys=[key]))

• Варіант за замовчуванням (без налаштування) підходить для одного процесу: stdio або рівно один робочий процес HTTP. Повторна спроба, що потрапляє на інший робочий процес, інший екземпляр за балансувальником навантаження чи на той самий сервер після перезапуску, запечатана ключем, якого цей процес не має, — клієнт отримує незмінну відмову, наведену нижче, і мусить почати процедуру спочатку.
• keys=[...] обов'язковий щоразу, коли повторна спроба може дістатися іншого екземпляра (uvicorn із кількома робочими процесами, HTTP за балансувальником) або має переживати перезапуски: кожен екземпляр перевіряє те, що випустив будь-який інший. Той самий механізм, лише ваш секрет замість згенерованого.
• Для власної криптографії, наприклад KMS чи наявного сервісу токенів, передайте RequestStateSecurity(codec=...) замість keys; контракт описано нижче в розділі Власна криптографія.

Що містить запечатаний токен

За замовчуванням чи з налаштуванням, у переданих даних requestState — це зашифрований автентифікований токен. Ваш код його ніколи не бачить: обробники й резолвери пишуть відкритий текст і читають відкритий текст (ctx.request_state); SDK запечатує на виході й перевіряє на вході. Окрім цілісності, кожен токен прив'язаний до:

• Часового вікна. Кожен раунд запечатує заново зі свіжим терміном дії, тож RequestStateSecurity(ttl=...) (за замовчуванням 600 секунд) обмежує час на роздуми в межах одного раунду, а не всю процедуру.
• Автентифікованого принципала. Коли запит містить токен доступу OAuth, який перевірив SDK, стан прив'язується до клієнта, видавця й суб'єкта токена: стан, випущений для одного користувача, не пройде перевірку в іншого, навіть якщо обидва користувачі мають спільний OAuth-клієнт. Верифікатор, що не надає суб'єкта, послаблює прив'язку до самої лише ідентичності клієнта, яку за URL-ідентифікаторів клієнтів поділяють усі користувачі цього клієнтського ПЗ. Коли автентифікація завершується поза SDK (на проксі попереду) або транспорт не автентифікований, прив'язувати немає до кого, і ця перевірка бездіяльна — хіба що RequestStateSecurity(bind_principal=...) надасть принципала з вашого власного сигналу ідентичності. Хай які компоненти надає ваш верифікатор токенів, він має надавати їх послідовно: верифікатор, що додає суб'єкта в одних запитах і пропускає в інших, змінює принципала посеред процедури, і незавершені раунди відхиляються.
• Початкового запиту. Метод, назва інструмента чи промпту (або URI ресурсу) і дайджест аргументів. Токен, відтворений для іншого інструмента, інших аргументів чи іншого методу, не проходить перевірку.
• Точного поставленого запитання. Кожна відповідь резолвера прикріплена до сформованого запитання, яке показали клієнту, — і в раунді, коли вона щойно надійшла, і коли записану відповідь використовують повторно пізніше. Розгорніть нову версію з переформульованим повідомленням чи зміненою схемою — і сервер перепитає, замість того щоб спожити застарілу відповідь. Та сама прив'язка працює й у зворотний бік: виводьте повідомлення з аргументів інструмента, а не з даних конкретного виклику. Повідомлення, побудоване з мітки часу чи поточного курсу, формується по-різному в кожному раунді, тож кожна записана відповідь здається застарілою, і сервер перепитує, доки ліміт раундів клієнта не завершить виклик.

Усе це — робота SDK, а не ваша і не кодека, якщо ви приносите власний.

Ротація ключів

keys[0] запечатує новий стан; перевіряє кожен ключ зі списку. Ротація без простою — це три фази, кожну з яких повністю розгортають перед наступною:

RequestStateSecurity(keys=[OLD, NEW])  # 1: every instance learns to verify NEW; OLD still mints
RequestStateSecurity(keys=[NEW, OLD])  # 2: NEW mints; in-flight OLD state keeps verifying
RequestStateSecurity(keys=[NEW])       # 3: one ttl after phase 2 is fully out, retire OLD

Ніколи не ставте новий ключ на випуск першим кроком: випуск під ключем, який котрийсь екземпляр іще не вміє перевіряти, обриває незавершені раунди посеред розгортання.

Ключі обмежені одним сервісом. Запечатаний конверт також несе назву сервера як твердження про аудиторію (audience claim), тож токен, випущений іншим сервісом, який випадково має той самий секрет, усе одно відхиляється. Твердження розрізняє сервіси рівно настільки, наскільки розрізняються назви, тож сервер з явно заданою політикою мусить мати справжню назву або задати RequestStateSecurity(audience=...) — безіменний викидає виняток під час створення. audience= також слугує навмисним багатосервісним топологіям, де один сервіс має приймати стан, випущений іншим. (На варіант за замовчуванням без налаштування це не поширюється: його ключ ніколи не покидає процес, тож твердженню про аудиторію нема чого додати.)

Власна криптографія

RequestStateSecurity(codec=...) приймає будь-що з методами seal(bytes) -> str і unseal(str) -> bytes, що викидає InvalidRequestState для будь-якого токена, якого не випускало. Класична форма — конвертне шифрування з KMS, коли ви один раз розгортаєте ключ даних під час запуску, а криптографію для кожного токена виконуєте локально:

import os

from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

from mcp.server import MCPServer
from mcp.server.mcpserver import InvalidRequestState, RequestStateSecurity

PREFIX = "kms1."  # format version; fed to GCM as associated data, so it is bound under the tag


def unwrap_data_key() -> bytes:
    """One KMS call at process start, kms.decrypt(CiphertextBlob=...); every token after that is local crypto."""
    return os.urandom(32)  # stand-in for the unwrapped 32-byte data key


class EnvelopeCodec:
    def __init__(self, data_key: bytes) -> None:
        self._aesgcm = AESGCM(data_key)

    def seal(self, payload: bytes) -> str:
        nonce = os.urandom(12)
        return PREFIX + (nonce + self._aesgcm.encrypt(nonce, payload, PREFIX.encode())).hex()

    def unseal(self, token: str) -> bytes:
        if not token.startswith(PREFIX):
            raise InvalidRequestState("unknown token format")
        body = token[len(PREFIX) :]
        try:
            raw = bytes.fromhex(body)
            if raw.hex() != body:  # only the exact string seal() produced verifies
                raise ValueError("non-canonical hex")
            return self._aesgcm.decrypt(raw[:12], raw[12:], PREFIX.encode())
        except (ValueError, InvalidTag) as exc:
            raise InvalidRequestState("token failed verification") from exc


mcp = MCPServer("Deployer", request_state_security=RequestStateSecurity(codec=EnvelopeCodec(unwrap_data_key())))

TTL, прив'язка до принципала й прив'язка до запиту — не робота кодека: SDK вписує їх у корисне навантаження перед seal і перевіряє заново після unseal для кожного кодека. Єдині обов'язки кодека — цілісність (підроблено — отже, виняток) і, в ідеалі, конфіденційність.

Коли перевірка не проходить

Кожна вхідна невдача — підробка, прострочення, відтворення для іншого запиту чи принципала або запечатування невідомим цьому серверу ключем — отримує ту саму відповідь:

{"code": -32602, "message": "Invalid or expired requestState"}

Одне незмінне повідомлення на всі причини, щоб передані дані ніколи не виказували, яка саме перевірка не пройшла; справжня причина потрапляє в лог сервера. Перевіряється кожен вхідний requestState у tools/call, prompts/get і resources/read, зокрема й той, що надходить для обробника, який ніколи не випускає стан. Найчастіша відмова на практиці — не зловмисник, а локальний для процесу ключ за замовчуванням, який зустрічає повторну спробу з-перед перезапуску чи з іншого екземпляра; клієнт починає процедуру спочатку, а keys=[...] — виправлення на випадок, коли це важливо.

Стан, зібраний вручну

request_state, який ви задаєте самі (повертаючи InputRequiredResult з функції інструмента, промпту чи шаблону ресурсу), запечатує й перевіряє той самий механізм, що й стан резолверів, без жодних змін у коді: пишете відкритий текст, читаєте відкритий текст, і всі наведені вище прив'язки діють.

Єдине, що SDK не може закріпити за вас навіть із налаштуванням, — це ідентичність запитання: він не знає, якому з ваших запитань належить відповідь у вашому стані. Якщо ви зберігаєте відповіді з ключами-запитаннями, додайте до стану власний ідентифікатор запитання й перевіряйте його під час повторної спроби.

Низькорівневий Server — це рівень без батарейок у комплекті: на відміну від MCPServer, тут нічого не запечатується, доки ви самі не додасте межу, а до того ваш request_state передається мережею рівно так, як написаний. Однорядкове ввімкнення показано на сторінці Низькорівневий Server.

Результат версії 2026-07-28

InputRequiredResult існує лише у версії протоколу 2026-07-28. Client(server) у пам'яті узгоджує її за вас; мережею її виявляє mode="auto". Після під'єднання client.protocol_version покаже, що саме ви отримали.

Підсумки

• У версії 2026-07-28 сервер, якому посеред виклику потрібні дані, повертає InputRequiredResult. Він ніколи не відкриває запит до клієнта.
• input_requests — це те, що йому потрібно. request_state — непрозорий токен відновлення, який читає лише сервер.
• Client виконує цикл повторних спроб за вас: зареєструйте elicitation_callback / sampling_callback / list_roots_callback — і call_tool повертає звичайний CallToolResult. Обмежує його input_required_max_rounds (за замовчуванням 10).
• Щоб перевіряти або зберігати раунди, використовуйте client.session.call_tool(..., allow_input_required=True) і ведіть цикл while isinstance(result, InputRequiredResult) власноруч.
• З @mcp.tool() цей результат за вас створює залежність, яка запитує користувача (Залежності); ручна форма — низькорівневий Server.
• Промпти й ресурси теж беруть участь: функція @mcp.prompt() або шаблонна @mcp.resource() сама повертає InputRequiredResult і читає ctx.input_responses під час повторної спроби.
• requestState повертається як дані, надані клієнтом, тож MCPServer за замовчуванням запечатує його — і стан резолверів, і стан, зібраний вручну, — локальним для процесу ключем; у розгортаннях із кількома екземплярами передавайте RequestStateSecurity(keys=[...]) (або власний кодек), щоб кожен екземпляр міг перевірити те, що випустив інший. Запечатування прив'язує кожен токен до часового вікна, початкового запиту й автентифікованого принципала — коли запит містить автентифікацію, яку перевірив SDK, або коли bind_principal= надає ваш власний сигнал ідентичності (Захист requestState).

Саме цей механізм замінює ініційоване сервером семплювання та решту зворотного каналу в стилі push; див. Застарілі можливості.