sse_utils.py

Source: src/sunholo/mcp/sse_utils.py

Functions

extract_sse_data(text: str) -> Optional[str]

Extract the data portion from an SSE response.

Args: text: SSE-formatted response text

Returns: The extracted data string, or None if not found

is_sse_response(text: str) -> bool

Check if a response is in SSE format.

Args: text: Response text to check

Returns: True if the response appears to be SSE format

parse_sse_response(text: str) -> Dict[str, Any]

Parse SSE-formatted response from MCP server.

FastMCP returns responses in SSE format when using HTTP transport, even with stateless_http=True. This function extracts the JSON data from the SSE format.

Args: text: Raw response text from MCP server

Returns: Parsed JSON data from the SSE response

Raises: ValueError: If the response cannot be parsed

Example: >>> response_text = 'event: message\ndata: {"jsonrpc": "2.0", "id": 1, "result": {...}}' >>> data = parse_sse_response(response_text) >>> print(data['result'])