Relevant source files

The following files were used as context for generating this wiki page: - [plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py](https://github.com/universal-tool-calling-protocol/python-utcp/blob/main/plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py) - [plugins/communication_protocols/http/src/utcp_http/sse_call_template.py](https://github.com/universal-tool-calling-protocol/python-utcp/blob/main/plugins/communication_protocols/http/src/utcp_http/sse_call_template.py) - [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py](https://github.com/universal-tool-calling-protocol/python-utcp/blob/main/plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py) - [plugins/communication_protocols/http/src/utcp_http/streamable_http_call_template.py](https://github.com/universal-tool-calling-protocol/python-utcp/blob/main/plugins/communication_protocols/http/src/utcp_http/streamable_http_call_template.py) - [README.md](https://github.com/universal-tool-calling-protocol/python-utcp/blob/main/README.md) - [plugins/communication_protocols/http/README.md](https://github.com/universal-tool-calling-protocol/python-utcp/blob/main/plugins/communication_protocols/http/README.md)

# Server-Sent Events (SSE) and Streaming The UTCP HTTP plugin provides robust support for real-time data and large response handling through two distinct streaming mechanisms: Server-Sent Events (SSE) and general-purpose Streamable HTTP. These features allow the `UtcpClient` to interact with services that push data continuously or return large payloads that are best processed in chunks. This document details the architecture and implementation of the `sse` and `streamable_http` communication protocols. Both are designed to be invoked via the `call_tool_streaming` method on the `UtcpClient`, providing an asynchronous generator that yields data as it arrives from the server. *Sources: [plugins/communication_protocols/http/README.md](), [README.md]()* ## Architecture Overview The streaming capabilities are implemented through two specialized `CommunicationProtocol` classes, each with a corresponding `CallTemplate`. * **`SseCommunicationProtocol`**: Handles connections to standard Server-Sent Events endpoints. It is configured using an `SseCallTemplate`. * **`StreamableHttpCommunicationProtocol`**: Manages generic HTTP streaming using chunked transfer encoding. It is configured using a `StreamableHttpCallTemplate`. The diagram below illustrates the relationship between these components. ```mermaid classDiagram direction LR class CommunicationProtocol { <> +register_manual() +deregister_manual() +call_tool() +call_tool_streaming() } class SseCommunicationProtocol { +call_tool_streaming() AsyncGenerator } class StreamableHttpCommunicationProtocol { +call_tool_streaming() AsyncGenerator -_process_http_stream() -_build_url_with_path_params() } class CallTemplate { <> name: str call_template_type: str } class SseCallTemplate { call_template_type: "sse" url: str event_type: str reconnect: bool } class StreamableHttpCallTemplate { call_template_type: "streamable_http" url: str http_method: str chunk_size: int } CommunicationProtocol <|-- SseCommunicationProtocol CommunicationProtocol <|-- StreamableHttpCommunicationProtocol CallTemplate <|-- SseCallTemplate CallTemplate <|-- StreamableHttpCallTemplate SseCommunicationProtocol ..> SseCallTemplate : uses StreamableHttpCommunicationProtocol ..> StreamableHttpCallTemplate : uses ``` *Sources: [plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py](), [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py](), [plugins/communication_protocols/http/src/utcp_http/sse_call_template.py](), [plugins/communication_protocols/http/src/utcp_http/streamable_http_call_template.py]()* ## Server-Sent Events (SSE) Protocol This protocol is designed specifically for interacting with endpoints that follow the W3C Server-Sent Events specification. It provides a persistent connection to a server, which can then push messages to the client. *Sources: [plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py]()* ### Configuration (`SseCallTemplate`) To define a tool that uses SSE, the `call_template_type` must be set to `"sse"`. The configuration is managed by the `SseCallTemplate` model. | Field | Type | Default | Description - | | `name` | `str` | - | A unique name for the manual call template. - | | `call_template_type` | `str` | `"sse"` | Must be `"sse"`. - | | `url` | `str` | - | The endpoint URL for the SSE stream. Must use `https://` or `http://localhost`. - | | `event_type` | `str` | `None` | The specific event type to listen for. If `None`, all messages are processed. - | | `reconnect` | `bool` | `True` | Whether to automatically attempt to reconnect if the connection is lost. - | | `retry_timeout` | `int` | `30000` | The delay in milliseconds before attempting to reconnect. - | | `auth` | `Auth` | `None` | Authentication configuration (ApiKey, Basic, or OAuth2). - | | `headers` | `dict` | `None` | A dictionary of static headers to include in the request. - | | `header_fields` | `list` | `None` | A list of tool argument keys whose values should be sent as headers. - | *Sources: [plugins/communication_protocols/http/src/utcp_http/sse_call_template.py](), [README.md:214-233]()* #### Example Configuration ```json { "name": "my_sse_stream", "call_template_type": "sse", "url": "https://api.example.com/events", "event_type": "message", "reconnect": true, "retry_timeout": 30000, "auth": { "auth_type": "basic", "username": "${USERNAME}", "password": "${PASSWORD}" }, "headers": { "X-Client-ID": "12345" } } ``` *Sources: [README.md:214-233]()* ### Execution Flow When `call_tool_streaming` is invoked for an SSE tool, the `SseCommunicationProtocol` initiates a connection to the specified URL. It uses the `aiohttp-sse-client` library to handle the SSE protocol details, including reconnection logic. The protocol then yields `ServerSentEvent` objects as they are received from the server. This diagram shows the sequence of operations for an SSE tool call. ```mermaid sequenceDiagram participant Client as UtcpClient participant SSE_Proto as SseCommunicationProtocol participant SSE_Client as aiohttp_sse_client participant Server as SSE Endpoint Client->>SSE_Proto: call_tool_streaming(tool_name, args) activate SSE_Proto Note over SSE_Proto: Apply auth and headers SSE_Proto->>SSE_Client: EventSource(url, headers, ...) activate SSE_Client SSE_Client->>Server: GET /events Server-->>SSE_Client: Connection established (text/event-stream) loop Events Stream Server-->>SSE_Client: data: {...} SSE_Client-->>SSE_Proto: yields ServerSentEvent SSE_Proto-->>Client: yields ServerSentEvent end deactivate SSE_Client deactivate SSE_Proto ``` *Sources: [plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py:165-201]()* ## Streamable HTTP Protocol The Streamable HTTP protocol is a more generic solution for handling any HTTP response that can be processed in chunks. It is not tied to a specific event format like SSE. This makes it suitable for downloading large files, consuming NDJSON streams, or any API that uses chunked transfer encoding. *Sources: [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py:24-29]()* ### Configuration (`StreamableHttpCallTemplate`) To use this protocol, set the `call_template_type` to `"streamable_http"`. The configuration is defined by the `StreamableHttpCallTemplate` model. | Field | Type | Default | Description - | | `name` | `str` | - | A unique name for the manual call template. - | | `call_template_type` | `str` | `"streamable_http"` | Must be `"streamable_http"`. The name `http_stream` is deprecated. - | | `url` | `str` | - | The endpoint URL. Supports path parameters like `/users/{user_id}`. Must use `https://` or `http://localhost`. - | | `http_method` | `str` | `"GET"` | The HTTP method to use (e.g., GET, POST). - | | `content_type` | `str` | `"application/octet-stream"` | The `Content-Type` header to set for POST/PUT requests with a body. - | | `chunk_size` | `int` | `4096` | The size of chunks (in bytes) to read from the stream for non-NDJSON content. - | | `timeout` | `int` | `60000` | The total request timeout in milliseconds. - | | `auth` | `Auth` | `None` | Authentication configuration (ApiKey, Basic, or OAuth2). - | | `body_field` | `str` | `None` | The key in tool arguments that contains the request body. - | | `header_fields` | `list` | `None` | A list of tool argument keys whose values should be sent as headers. - | *Sources: [plugins/communication_protocols/http/src/utcp_http/streamable_http_call_template.py](), [README.md:236-250]()* #### Example Configuration ```json { "name": "streaming_data_source", "call_template_type": "streamable_http", "url": "https://api.example.com/stream", "http_method": "POST", "content_type": "application/octet-stream", "chunk_size": 4096, "timeout": 60000, "body_field": "data" } ``` *Sources: [README.md:236-250]()* ### Execution Flow and Chunk Processing The `StreamableHttpCommunicationProtocol` uses `aiohttp` to make a request. The core logic resides in the `_process_http_stream` method, which inspects the `Content-Type` header of the response to determine how to process and yield chunks. This flowchart illustrates the chunk processing logic: ```mermaid flowchart TD A[Start _process_http_stream] --> B{Get Content-Type}; B --> C{application/x-ndjson?}; C -- Yes --> D[Iterate over lines]; D --> E[Yield json.loads(line)]; E --> D; C -- No --> F{application/octet-stream?}; F -- Yes --> G[Iterate chunked by chunk_size]; G --> H[Yield chunk]; H --> G; F -- No --> I{application/json?}; I -- Yes --> J[Buffer entire response]; J --> K[Yield json.loads(buffer)]; I -- No --> L[Default: Binary Stream]; L --> M[Iterate chunked by chunk_size]; M --> N[Yield chunk]; N --> M; ``` This logic allows the protocol to intelligently handle different streaming formats, such as yielding individual JSON objects for NDJSON streams or raw bytes for binary downloads. *Sources: [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py:313-350]()* ### Path Parameter Substitution The `StreamableHttpCommunicationProtocol` (and its non-streaming counterpart) supports substituting parts of the URL with values from tool arguments. This is achieved by placing placeholders like `{param_name}` in the `url` field of the call template. The `_build_url_with_path_params` method handles this substitution. - It finds all `{...}` placeholders in the URL. - For each placeholder, it looks for a matching key in the `tool_args`. - If found, it URL-encodes the value and replaces the placeholder. - The argument is then removed from `tool_args` to prevent it from being used as a query parameter. - If a required path parameter is missing from the arguments, a `ValueError` is raised. *Sources: [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py:381-411]()*, *[plugins/communication_protocols/http/tests/test_http_communication_protocol.py:306-343]()* ## Authentication Both streaming protocols share the same authentication handling logic, supporting API Key, Basic, and OAuth2 authentication methods. The `_apply_auth` method configures headers, query parameters, or `aiohttp` auth objects based on the `auth` configuration in the call template. OAuth2 is handled separately by the `_handle_oauth2` method due to its asynchronous nature. *Sources: [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py:41-72](), [plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py:47-78]()* ### OAuth2 Client Credentials Flow The `_handle_oauth2` method implements the client credentials grant type. It first attempts to get a token by sending credentials in the request body. If that fails, it retries by sending the credentials as a `Basic` authentication header, which is a common variation. Acquired tokens are cached in memory for reuse. This diagram shows the OAuth2 token acquisition flow: ```mermaid flowchart TD A[Start _handle_oauth2] --> B{Token in cache?}; B -- Yes --> C[Return cached token]; B -- No --> D[Attempt 1: Send credentials in POST body]; D --> E{Request successful?}; E -- Yes --> F[Cache and return token]; E -- No --> G[Attempt 2: Send credentials as Basic Auth header]; G --> H{Request successful?}; H -- Yes --> F; H -- No --> I[Raise exception]; ``` *Sources: [plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py:352-379](), [plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py:203-230]()* ## Usage: `call_tool` vs. `call_tool_streaming` While streaming protocols are designed to be used with `call_tool_streaming`, they also implement the standard `call_tool` method. The `call_tool` implementation for these protocols simply consumes the entire asynchronous stream from `call_tool_streaming` and aggregates the chunks into a single result (either a list of chunks or a single bytes object). ```python # From StreamableHttpCommunicationProtocol async def call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) ->