Skip to main content
Version: Next

utcp_client

File: core/src/utcp/utcp_client.py

class UtcpClient​

Documentation

Abstract interface for UTCP client implementations.

Defines the core contract for UTCP clients, including CallTemplate management, tool execution, search capabilities, and variable handling. This interface allows for different client implementations while maintaining consistency.

Methods:​

async create(cls, root_dir: Optional[str], config: Optional[Union[str, Dict[str, Any], 'UtcpClientConfig']]) -> 'UtcpClient'

Create a new instance of UtcpClient.

Args

  • root_dir: The root directory for the client to resolve relative paths from. Defaults to the current working directory.
  • config: The configuration for the client. Can be a path to a configuration file, a dictionary, or UtcpClientConfig object.
  • tool_repository: The tool repository to use. Defaults to InMemToolRepository.
  • search_strategy: The tool search strategy to use. Defaults to TagSearchStrategy.

Returns

A new instance of UtcpClient.

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

Register a tool CallTemplate and its tools.

Args

Returns

A RegisterManualResult object containing the registered CallTemplate and its tools.

async register_manuals(self, manual_call_templates: List[CallTemplate]) -> List[RegisterManualResult]

Register multiple tool CallTemplates and their tools.

Args

  • manual_call_templates: List of CallTemplates to register.

Returns

A list of RegisterManualResult objects containing the registered CallTemplates and their tools. Order is not preserved.

async deregister_manual(self, manual_call_template_name: str) -> bool

Deregister a tool CallTemplate.

Args

  • manual_call_template_name: The name of the CallTemplate to deregister.

Returns

True if the CallTemplate was deregistered, False otherwise.

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

Call a tool.

Args

  • tool_name: The name of the tool to call.
  • tool_args: The arguments to pass to the tool.

Returns

The result of the tool call.

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

Call a tool streamingly.

Args

  • tool_name: The name of the tool to call.
  • tool_args: The arguments to pass to the tool.

Returns

An async generator that yields the result of the tool call.

async search_tools(self, query: str, limit: int, any_of_tags_required: Optional[List[str]]) -> List[Tool]

Search for tools relevant to the query.

Args

  • query: The search query.
  • limit: The maximum number of tools to return. 0 for no limit.
  • any_of_tags_required: Optional list of tags where one of them must be present in the tool's tags

Returns

A list of tools that match the search query.

async get_required_variables_for_manual_and_tools(self, manual_call_template: CallTemplate) -> List[str]

Get the required variables for a manual CallTemplate and its tools.

Args

Returns

A list of required variables for the manual CallTemplate and its tools.

async get_required_variables_for_registered_tool(self, tool_name: str) -> List[str]

Get the required variables for a registered tool.

Args

  • tool_name: The name of a registered tool.

Returns

A list of required variables for the tool.

async close(self) -> None

Close the protocol instances this client created and release their resources.

Closes every instance the client obtained from CommunicationProtocol.communication_protocol_factories. Instances from CommunicationProtocol.communication_protocols are shared by every client in the process and are left running.

Every owned instance is closed even if one of them fails to close; the failures are then raised together as a UtcpProtocolCloseError.