Validate the session header. Raises ``ValueError`` when the header is missing. Returns ``True`` when the session was stale and transparently recovered, or ``False`` when the session is valid.
(method: str)
| 80 | # -- Streamable HTTP endpoint -------------------------------------------- |
| 81 | |
| 82 | def _check_session(method: str) -> bool: |
| 83 | """Validate the session header. |
| 84 | |
| 85 | Raises ``ValueError`` when the header is missing. |
| 86 | Returns ``True`` when the session was stale and transparently |
| 87 | recovered, or ``False`` when the session is valid. |
| 88 | """ |
| 89 | nonlocal _session_id |
| 90 | adapter = app.backend.request_adapter() |
| 91 | if method == "initialize": |
| 92 | _session_id = _get_or_create_session_id() |
| 93 | return False |
| 94 | client_session_id = adapter.headers.get("Mcp-Session-Id") |
| 95 | if client_session_id and _is_session_stale(client_session_id): |
| 96 | _session_id = _get_or_create_session_id() |
| 97 | logger.debug("MCP session recovered: %s", _session_id) |
| 98 | return True |
| 99 | return False |
| 100 | |
| 101 | def _json_response(*messages: dict): |
| 102 | """Wrap one or more JSON-RPC messages in a response. |
no test coverage detected
searching dependent graphs…