UTCP工具

Install the core package in editable mode with dev dependencies

pip install -e "core[dev]"

Ensure you have installed all dev dependencies

python -m pytest

To run tests for a specific package (e.g., the core library):

To run tests for a specific plugin (e.g., HTTP):

To run tests with coverage:

Installation

Install the core library and any required protocol plugins:

```bash

Install core + HTTP plugin (most common)

pip install utcp utcp-http

Install additional plugins as needed

pip install utcp-cli utcp-mcp utcp-text ```

Install a specific protocol plugin in editable mode

pip install -e plugins/communication_protocols/http ```

Build

The build process now involves building each package (core and plugins) separately if needed, though they are published to PyPI independently.

1. Create and activate a virtual environment.
2. Install build dependencies: pip install build.
3. Navigate to the package directory (e.g., cd core).
4. Run the build: python -m build.
5. The distributable files (.whl and .tar.gz) will be in the dist/ directory.

Quick Start

Basic Usage

```python from utcp.utcp_client import UtcpClient

Migration Guide from 0.x to 1.0.0

Version 1.0.0 introduces several breaking changes. Follow these steps to migrate your project.

1. Update Dependencies: Install the new utcp core package and the specific protocol plugins you use (e.g., utcp-http, utcp-cli). 2. Configuration: Configuration Object: UtcpClient is initialized with a UtcpClientConfig object, dict or a path to a JSON file containing the configuration. Manual Call Templates: The providers_file_path option is removed. Instead of a file path, you now provide a list of manual_call_templates directly within the UtcpClientConfig. Terminology: The term provider has been replaced with call_template, and provider_type is now call_template_type. Streamable HTTP: The call_template_type http_stream has been renamed to streamable_http. 3. Update Imports: Change your imports to reflect the new modular structure. For example, from utcp.client.transport_interfaces.http_transport import HttpProvider becomes from utcp_http.http_call_template import HttpCallTemplate. 4. Tool Search: If you were using the default search, the new strategy is TagAndDescriptionWordMatchStrategy. This is the new default and requires no changes unless you were implementing a custom strategy. 5. Tool Naming: Tool names are now namespaced as manual_name.tool_name. The client handles this automatically. 6. Variable Substitution Namespacing: Variables that are substituted in different call_templates, are first namespaced with the name of the manual with the _ duplicated. So a key in a tool call template called API_KEY from the manual manual_1 would be converted to manual__1_API_KEY.

Usage Examples

3. Full examples

You can find full examples in the examples repository.

Call Template Configuration Examples

Configuration examples for each protocol. Remember to replace provider_type with call_template_type.

Quick Start with OpenAPI

```python from utcp_http.openapi_converter import OpenApiConverter import aiohttp

JSON Configuration

{
  "name": "my_api",
  "call_template_type": "http",
  "url": "https://api.example.com/utcp",
  "allowed_communication_protocols": ["http", "cli", "mcp"]
}

Or use UTCP Client configuration for automatic detection

from utcp.utcp_client import UtcpClient

client = await UtcpClient.create(config={ "manual_call_templates": [{ "name": "github", "call_template_type": "http", "url": "https://api.github.com/openapi.json", "auth_tools": { # Authentication for generated tools requiring auth "auth_type": "api_key", "api_key": "Bearer ${GITHUB_TOKEN}", "var_name": "Authorization", "location": "header" } }] }) ```

Create client with HTTP API

client = await UtcpClient.create(config={ "manual_call_templates": [{ "name": "my_api", "call_template_type": "http", "url": "https://api.example.com/utcp" }] })

The discovery endpoint returns the tool manual

@app.get("/utcp") def utcp_discovery(): return UtcpManual.create_from_decorators(manual_version="1.0.0")

The actual tool endpoint

@utcp_tool(tool_call_template=HttpCallTemplate( name="get_weather", url=f"https://example.com/api/weather", http_method="GET" ), tags=["weather"]) @app.get("/api/weather") def get_weather(location: str): return {"temperature": 22.5, "conditions": "Sunny"}

No UTCP dependencies server version:

app = FastAPI()

The discovery endpoint returns the tool manual

@app.get("/utcp") def utcp_discovery(): return { "manual_version": "1.0.0", "utcp_version": "1.0.2", "tools": [ { "name": "get_weather", "description": "Get current weather for a location", "tags": ["weather"], "inputs": { "type": "object", "properties": { "location": {"type": "string"} } }, "outputs": { "type": "object", "properties": { "temperature": {"type": "number"}, "conditions": {"type": "string"} } }, "tool_call_template": { "call_template_type": "http", "url": "https://example.com/api/weather", "http_method": "GET" } } ] }

The actual tool endpoint

@app.get("/api/weather") def get_weather(location: str): return {"temperature": 22.5, "conditions": "Sunny"} ```

CLI Call Template

{
  "name": "multi_step_cli_tool",
  "call_template_type": "cli", // Required
  "commands": [ // Required - sequential command execution
    {
      "command": "git clone UTCP_ARG_repo_url_UTCP_END temp_repo",
      "append_to_final_output": false
    },
    {
      "command": "cd temp_repo && find . -name '*.py' | wc -l"
      // Last command output returned by default
    }
  ],
  "env_vars": { // Optional
    "GIT_AUTHOR_NAME": "UTCP Bot",
    "API_KEY": "${MY_API_KEY}"
  },
  "working_dir": "/tmp", // Optional
  "auth": null // Optional (always null for CLI)
}

CLI Protocol Features: - Multi-command execution: Commands run sequentially in single subprocess - Cross-platform: PowerShell on Windows, Bash on Unix/Linux/macOS - State preservation: Directory changes (cd) persist between commands - Argument placeholders: UTCP_ARG_argname_UTCP_END format - Output referencing: Access previous outputs with $CMD_0_OUTPUT, $CMD_1_OUTPUT - Flexible output control: Choose which command outputs to include in final result

This manual can register/call both HTTP and CLI tools

multi_protocol_manual = HttpCallTemplate( name="flexible_manual", call_template_type="http", url="https://api.example.com/utcp", allowed_communication_protocols=["http", "cli"] # Explicitly allow both ) ```

Raises ValueError: Tool 'my_api.some_cli_tool' uses communication protocol 'cli'

which is not allowed by manual 'my_api'. Allowed protocols: ['http']

await client.call_tool("my_api.some_cli_tool", {"arg": "value"}) ```

OpenAPI Ingestion - Zero Infrastructure Tool Integration

🚀 Transform any existing REST API into UTCP tools without server modifications!

UTCP's OpenAPI ingestion feature automatically converts OpenAPI 2.0/3.0 specifications into UTCP tools, enabling AI agents to interact with existing APIs directly - no wrapper servers, no API changes, no additional infrastructure required.

Convert any OpenAPI spec to UTCP tools

async def convert_api(): async with aiohttp.ClientSession() as session: async with session.get("https://api.github.com/openapi.json") as response: openapi_spec = await response.json() converter = OpenApiConverter(openapi_spec) manual = converter.convert() print(f"Generated {len(manual.tools)} tools from GitHub API!") return manual

Core Package (`utcp`)

The core/ directory contains the foundational components: - Data Models: Pydantic models for Tool, CallTemplate, UtcpManual, and Auth - Client Interface: Main UtcpClient for tool interaction - Plugin System: Extensible interfaces for protocols, repositories, and search - Default Implementations: Built-in tool storage and search strategies

Protocol Plugins

UTCP supports multiple communication protocols through dedicated plugins:

For development, you can install the packages in editable mode from the cloned repository:

```bash

Protocol Lifetime: Shared vs. Per-Client Instances

A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to.

An instance registered with register_communication_protocol is shared by every UtcpClient in the process, and so is any state it keeps. That is right for state that is meant to be shared — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it.

A protocol whose state must not be shared between clients registers a factory instead: live sessions or connections keyed per manual, child processes — anything one client's use or close() would take away from another. Each UtcpClient calls the factory once — at creation, or on first use if it was registered later — so each client gets its own instance, its own connections, and its own teardown on close():

from utcp.plugins.discovery import register_communication_protocol_factory

register_communication_protocol_factory("custom_type", CustomCommunicationProtocol)

This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing.

utcp-mcp and utcp-websocket register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. utcp-http (HTTP, SSE, streamable HTTP) and utcp-gql stay shared instances: the only state they keep is an OAuth token cache, which is meant to be reused across clients; every request opens its own session.

client.close() closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as UtcpProtocolCloseError. If UtcpClient.create fails after creating them, they are closed the same way before the error is re-raised.