Function _plan_outbound

src/mcp/shared/jsonrpc_dispatcher.py:201–226

Choose the outbound `SessionMessage.metadata` and the abandon-cancellation policy. `related_request_id` wins over resumption hints (they are dropped). Only hints that actually reach the transport suppress the courtesy cancel - a request that is neither resumable nor cancelled would leak

(related_request_id: RequestId | None, opts: CallOptions | None)

def _plan_outbound(related_request_id: RequestId | None, opts: CallOptions | None) -> _OutboundPlan:
    """Choose the outbound `SessionMessage.metadata` and the abandon-cancellation policy.

    `related_request_id` wins over resumption hints (they are dropped). Only
    hints that actually reach the transport suppress the courtesy cancel - a
    request that is neither resumable nor cancelled would leak the peer's work.
    """
    opts = opts or {}
    cancel_on_abandon = opts.get("cancel_on_abandon", True)
    token = opts.get("resumption_token")
    on_token = opts.get("on_resumption_token")
    headers = opts.get("headers")
    if related_request_id is not None:
        if token is not None or on_token is not None:
            logger.debug(
                "dropping resumption hints: related_request_id %r takes precedence on metadata", related_request_id
            )
        return _OutboundPlan(ServerMessageMetadata(related_request_id=related_request_id), cancel_on_abandon)
    if token is not None or on_token is not None:
        return _OutboundPlan(
            ClientMessageMetadata(resumption_token=token, on_resumption_token_update=on_token, headers=headers),
            cancel_on_abandon=False,
        )
    if headers:
        return _OutboundPlan(ClientMessageMetadata(headers=headers), cancel_on_abandon)
    return _OutboundPlan(None, cancel_on_abandon)


class JSONRPCDispatcher(Dispatcher[TransportT]):