| 68 | |
| 69 | |
| 70 | def _make_modern_stamp( |
| 71 | protocol_version: str, client_info: dict[str, Any], capabilities: dict[str, Any] |
| 72 | ) -> Callable[[dict[str, Any], CallOptions], None]: |
| 73 | def stamp(data: dict[str, Any], opts: CallOptions) -> None: |
| 74 | params = data.setdefault("params", {}) |
| 75 | meta = params.setdefault("_meta", {}) |
| 76 | meta[PROTOCOL_VERSION_META_KEY] = protocol_version |
| 77 | meta[CLIENT_INFO_META_KEY] = client_info |
| 78 | meta[CLIENT_CAPABILITIES_META_KEY] = capabilities |
| 79 | opts["cancel_on_abandon"] = False |
| 80 | headers = opts.setdefault("headers", {}) |
| 81 | headers[MCP_PROTOCOL_VERSION_HEADER] = protocol_version |
| 82 | headers[MCP_METHOD_HEADER] = data["method"] |
| 83 | name_key = NAME_BEARING_METHODS.get(data["method"]) |
| 84 | if name_key is not None and isinstance(name := params.get(name_key), str): |
| 85 | headers[MCP_NAME_HEADER] = encode_header_value(name) |
| 86 | |
| 87 | return stamp |
| 88 | |
| 89 | |
| 90 | ReceiveResultT = TypeVar("ReceiveResultT", bound=BaseModel) |