Skip to main content
Version: Next

http_communication_protocol

File: plugins/communication_protocols/http/src/utcp_http/http_communication_protocol.py

class HttpCommunicationProtocol (CommunicationProtocol)

Documentation

HTTP communication protocol implementation for UTCP client.

Handles communication with HTTP-based tool providers, supporting various authentication methods, URL path parameters, and automatic tool discovery. Enforces security by requiring HTTPS or localhost connections.

Features

  • RESTful API communication with configurable HTTP methods
  • URL path parameter substitution from tool arguments
  • Tool discovery from UTCP manuals, OpenAPI specs, and YAML
  • Request body and header field mapping from tool arguments
  • OAuth2 token caching and automatic refresh
  • Security validation of connection URLs

Attributes

  • _session: Optional aiohttp ClientSession for connection reuse.
  • _oauth_tokens: Cache of OAuth2 tokens by client_id.
  • _log: Logger function for debugging and error reporting.

Methods:

async register_manual(self, caller, manual_call_template: CallTemplate) -> RegisterManualResult

Register a manual and its tools.

Args

  • caller: The UTCP client that is calling this method.
  • manual_call_template: The call template of the manual to register.

Returns

RegisterManualResult object containing the call template and manual.

async deregister_manual(self, caller, manual_call_template: CallTemplate) -> None

No method documentation available

async call_tool(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> Any

Execute a tool call through this transport.

Args

  • caller: The UTCP client that is calling this method.
  • tool_name: Name of the tool to call (may include provider prefix).
  • tool_args: Dictionary of arguments to pass to the tool.
  • tool_call_template: Call template of the tool to call.

Returns

The tool's response, with type depending on the tool's output schema.

async call_tool_streaming(self, caller, tool_name: str, tool_args: Dict[str, Any], tool_call_template: CallTemplate) -> AsyncGenerator[Any, None]

Execute a tool call through this transport streamingly.

Args

  • caller: The UTCP client that is calling this method.
  • tool_name: Name of the tool to call (may include provider prefix).
  • tool_args: Dictionary of arguments to pass to the tool.
  • tool_call_template: Call template of the tool to call.

Returns

An async generator that yields the tool's response.