Universal Tool Calling Protocol (UTCP)
Introduction
The Universal Tool Calling Protocol (UTCP) is a secure, scalable standard for defining and interacting with tools across a wide variety of communication protocols. UTCP 1.0.0 introduces a modular core with a plugin-based architecture, making it more extensible, testable, and easier to package.
In contrast to other protocols, UTCP places a strong emphasis on:
- Scalability: UTCP is designed to handle a large number of tools and providers without compromising performance.
- Extensibility: A pluggable architecture allows developers to easily add new communication protocols, tool storage mechanisms, and search strategies without modifying the core library.
- Interoperability: With a growing ecosystem of protocol plugins (including HTTP, SSE, CLI, and more), UTCP can integrate with almost any existing service or infrastructure.
- Ease of Use: The protocol is built on simple, well-defined Pydantic models, making it easy for developers to implement and use.
Repository Structure
This repository contains the complete UTCP Python implementation:
core/- Coreutcppackage with foundational components (README)plugins/communication_protocols/- Protocol-specific plugins:
Architecture Overview
UTCP uses a modular architecture with a core library and protocol plugins:
Core Package (utcp)
The core/ directory contains the foundational components:
- Data Models: Pydantic models for
Tool,CallTemplate,UtcpManual, andAuth - Client Interface: Main
UtcpClientfor tool interaction - Plugin System: Extensible interfaces for protocols, repositories, and search
- Default Implementations: Built-in tool storage and search strategies
Quick Start
Installation
Install the core library and any required protocol plugins:
# Install core + HTTP plugin (most common)
pip install utcp utcp-http
# Install additional plugins as needed
pip install utcp-cli utcp-mcp utcp-text
Basic Usage
from utcp.utcp_client import UtcpClient
# 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"
}]
})
# Call a tool
result = await client.call_tool("my_api.get_data", {"id": "123"})
# Release the connections this client opened
await client.close()
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.
Protocol Plugins
UTCP supports multiple communication protocols through dedicated plugins:
For development, you can install the packages in editable mode from the cloned repository:
# Clone the repository
git clone https://github.com/universal-tool-calling-protocol/python-utcp.git
cd python-utcp
# Install the core package in editable mode with dev dependencies
pip install -e "core[dev]"
# Install a specific protocol plugin in editable mode
pip install -e plugins/communication_protocols/http
Migration Guide from 0.x to 1.0.0
Version 1.0.0 introduces several breaking changes. Follow these steps to migrate your project.
- Update Dependencies: Install the new
utcpcore package and the specific protocol plugins you use (e.g.,utcp-http,utcp-cli). - Configuration:
- Configuration Object:
UtcpClientis initialized with aUtcpClientConfigobject, dict or a path to a JSON file containing the configuration. - Manual Call Templates: The
providers_file_pathoption is removed. Instead of a file path, you now provide a list ofmanual_call_templatesdirectly within theUtcpClientConfig. - Terminology: The term
providerhas been replaced withcall_template, andprovider_typeis nowcall_template_type. - Streamable HTTP: The
call_template_typehttp_streamhas been renamed tostreamable_http.
- Configuration Object:
- Update Imports: Change your imports to reflect the new modular structure. For example,
from utcp.client.transport_interfaces.http_transport import HttpProviderbecomesfrom utcp_http.http_call_template import HttpCallTemplate. - 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. - Tool Naming: Tool names are now namespaced as
manual_name.tool_name. The client handles this automatically. - 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 calledAPI_KEYfrom the manualmanual_1would be converted tomanual__1_API_KEY.
Usage Examples
1. Using the UTCP Client
config.json (Optional)
You can define a comprehensive client configuration in a JSON file. All of these fields are optional.
{
"variables": {
"openlibrary_URL": "https://openlibrary.org/static/openapi.json"
},
"load_variables_from": [
{
"variable_loader_type": "dotenv",
"env_file_path": ".env"
}
],
"tool_repository": {
"tool_repository_type": "in_memory"
},
"tool_search_strategy": {
"tool_search_strategy_type": "tag_and_description_word_match"
},
"manual_call_templates": [
{
"name": "openlibrary",
"call_template_type": "http",
"http_method": "GET",
"url": "${URL}",
"content_type": "application/json"
},
],
"post_processing": [
{
"tool_post_processor_type": "filter_dict",
"only_include_keys": ["name", "key"],
"only_include_tools": ["openlibrary.read_search_authors_json_search_authors_json_get"]
}
]
}
client.py
import asyncio
from utcp.utcp_client import UtcpClient
from utcp.data.utcp_client_config import UtcpClientConfig
async def main():
# The UtcpClient can be created with a config file path, a dict, or a UtcpClientConfig object.
# Option 1: Initialize from a config file path
# client_from_file = await UtcpClient.create(config="./config.json")
# Option 2: Initialize from a dictionary
client_from_dict = await UtcpClient.create(config={
"variables": {
"openlibrary_URL": "https://openlibrary.org/static/openapi.json"
},
"load_variables_from": [
{
"variable_loader_type": "dotenv",
"env_file_path": ".env"
}
],
"tool_repository": {
"tool_repository_type": "in_memory"
},
"tool_search_strategy": {
"tool_search_strategy_type": "tag_and_description_word_match"
},
"manual_call_templates": [
{
"name": "openlibrary",
"call_template_type": "http",
"http_method": "GET",
"url": "${URL}",
"content_type": "application/json"
}
],
"post_processing": [
{
"tool_post_processor_type": "filter_dict",
"only_include_keys": ["name", "key"],
"only_include_tools": ["openlibrary.read_search_authors_json_search_authors_json_get"]
}
]
})
# Option 3: Initialize with a full-featured UtcpClientConfig object
from utcp_http.http_call_template import HttpCallTemplate
from utcp.data.variable_loader import VariableLoaderSerializer
from utcp.interfaces.tool_post_processor import ToolPostProcessorConfigSerializer
config_obj = UtcpClientConfig(
variables={"openlibrary_URL": "https://openlibrary.org/static/openapi.json"},
load_variables_from=[
VariableLoaderSerializer().validate_dict({
"variable_loader_type": "dotenv", "env_file_path": ".env"
})
],
manual_call_templates=[
HttpCallTemplate(
name="openlibrary",
call_template_type="http",
http_method="GET",
url="${URL}",
content_type="application/json"
)
],
post_processing=[
ToolPostProcessorConfigSerializer().validate_dict({
"tool_post_processor_type": "filter_dict",
"only_include_keys": ["name", "key"],
"only_include_tools": ["openlibrary.read_search_authors_json_search_authors_json_get"]
})
]
)
client = await UtcpClient.create(config=config_obj)
# Call a tool. The name is namespaced: `manual_name.tool_name`
result = await client.call_tool(
tool_name="openlibrary.read_search_authors_json_search_authors_json_get",
tool_args={"q": "J. K. Rowling"}
)
print(result)
if __name__ == "__main__":
asyncio.run(main())
2. Providing a UTCP Manual
A UTCPManual describes the tools you offer. The key change is replacing tool_provider with tool_call_template.
server.py
UTCP decorator version:
from fastapi import FastAPI
from utcp_http.http_call_template import HttpCallTemplate
from utcp.data.utcp_manual import UtcpManual
from utcp.python_specific_tooling.tool_decorator import utcp_tool
app = FastAPI()
# 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:
from fastapi import FastAPI
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"}
3. Full examples
You can find full examples in the examples repository.
Protocol Specification
UtcpManual and Tool Models
The tool_provider object inside a Tool has been replaced by tool_call_template.
{
"manual_version": "string",
"utcp_version": "string",
"tools": [
{
"name": "string",
"description": "string",
"inputs": { ... },
"outputs": { ... },
"tags": ["string"],
"tool_call_template": {
"call_template_type": "http",
"url": "https://...",
"http_method": "GET"
}
}
]
}
Call Template Configuration Examples
Configuration examples for each protocol. Remember to replace provider_type with call_template_type.
HTTP Call Template
{
"name": "my_rest_api",
"call_template_type": "http", // Required
"url": "https://api.example.com/users/{user_id}", // Required
"http_method": "POST", // Required, default: "GET"
"content_type": "application/json", // Optional, default: "application/json"
"allowed_communication_protocols": ["http"], // Optional, defaults to [call_template_type]. Restricts which protocols tools can use.
"auth": { // Optional, authentication for the HTTP request (example using ApiKeyAuth for Bearer token)
"auth_type": "api_key",
"api_key": "Bearer $API_KEY", // Required
"var_name": "Authorization", // Optional, default: "X-Api-Key"
"location": "header" // Optional, default: "header"
},
"auth_tools": { // Optional, authentication for converted tools, if this call template points to an openapi spec that should be automatically converted to a utcp manual (applied only to endpoints requiring auth per OpenAPI spec)
"auth_type": "api_key",
"api_key": "Bearer $TOOL_API_KEY", // Required
"var_name": "Authorization", // Optional, default: "X-Api-Key"
"location": "header" // Optional, default: "header"
},
"headers": { // Optional
"X-Custom-Header": "value"
},
"body_field": "body", // Optional, default: "body"
"header_fields": ["user_id"] // Optional
}
SSE (Server-Sent Events) Call Template
{
"name": "my_sse_stream",
"call_template_type": "sse", // Required
"url": "https://api.example.com/events", // Required
"event_type": "message", // Optional
"reconnect": true, // Optional, default: true
"retry_timeout": 30000, // Optional, default: 30000 (ms)
"auth": { // Optional, example using BasicAuth
"auth_type": "basic",
"username": "${USERNAME}", // Required
"password": "${PASSWORD}" // Required
},
"headers": { // Optional
"X-Client-ID": "12345"
},
"body_field": null, // Optional
"header_fields": [] // Optional
}
Streamable HTTP Call Template
Note the name change from http_stream to streamable_http.
{
"name": "streaming_data_source",
"call_template_type": "streamable_http", // Required
"url": "https://api.example.com/stream", // Required
"http_method": "POST", // Optional, default: "GET"
"content_type": "application/octet-stream", // Optional, default: "application/octet-stream"
"chunk_size": 4096, // Optional, default: 4096
"timeout": 60000, // Optional, default: 60000 (ms)
"auth": null, // Optional
"headers": {}, // Optional
"body_field": "data", // Optional
"header_fields": [] // Optional
}
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_ENDformat - Output referencing: Access previous outputs with
$CMD_0_OUTPUT,$CMD_1_OUTPUT - Flexible output control: Choose which command outputs to include in final result
Text Call Template
{
"name": "my_text_manual",
"call_template_type": "text", // Required
"file_path": "./manuals/my_manual.json", // Required
"auth": null, // Optional (always null for Text)
"auth_tools": { // Optional, authentication for generated tools from OpenAPI specs
"auth_type": "api_key",
"api_key": "Bearer ${API_TOKEN}",
"var_name": "Authorization",
"location": "header"
}
}
MCP (Model Context Protocol) Call Template
{
"name": "my_mcp_server",
"call_template_type": "mcp", // Required
"config": { // Required
"mcpServers": {
"server_name": {
"transport": "stdio",
"command": ["python", "-m", "my_mcp_server"]
}
}
},
"auth": { // Optional, example using OAuth2
"auth_type": "oauth2",
"token_url": "https://auth.example.com/token", // Required
"client_id": "${CLIENT_ID}", // Required
"client_secret": "${CLIENT_SECRET}", // Required
"scope": "read:tools" // Optional
}
}
Security: Protocol Restrictions
UTCP provides fine-grained control over which communication protocols each manual can use through the allowed_communication_protocols field. This prevents potentially dangerous protocol escalation (e.g., an HTTP-based manual accidentally calling CLI tools).
Default Behavior (Secure by Default)
When allowed_communication_protocols is not set or is empty, a manual can only register and call tools that use the same protocol type as the manual itself:
from utcp_http.http_call_template import HttpCallTemplate
# This manual can ONLY register/call HTTP tools (default restriction)
http_manual = HttpCallTemplate(
name="my_api",
call_template_type="http",
url="https://api.example.com/utcp"
# allowed_communication_protocols not set → defaults to ["http"]
)
Allowing Multiple Protocols
To allow a manual to work with tools from multiple protocols, explicitly set allowed_communication_protocols:
from utcp_http.http_call_template import HttpCallTemplate
# 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
)
JSON Configuration
{
"name": "my_api",
"call_template_type": "http",
"url": "https://api.example.com/utcp",
"allowed_communication_protocols": ["http", "cli", "mcp"]
}
Behavior Summary
Registration Filtering
During register_manual(), tools that don't match the allowed protocols are automatically filtered out with a warning:
WARNING - Tool 'dangerous_tool' uses communication protocol 'cli' which is not in
allowed protocols ['http'] for manual 'my_api'. Tool will not be registered.
Call-Time Validation
Even if a tool somehow exists in the repository, calling it will fail if its protocol is not allowed:
# 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"})
Testing
The testing structure has been updated to reflect the new core/plugin split.
Running Tests
To run all tests for the core library and all plugins:
# Ensure you have installed all dev dependencies
python -m pytest
To run tests for a specific package (e.g., the core library):
python -m pytest core/tests/
To run tests for a specific plugin (e.g., HTTP):
python -m pytest plugins/communication_protocols/http/tests/ -v
To run tests with coverage:
python -m pytest --cov=utcp --cov-report=xml
Build
The build process now involves building each package (core and plugins) separately if needed, though they are published to PyPI independently.
- Create and activate a virtual environment.
- Install build dependencies:
pip install build. - Navigate to the package directory (e.g.,
cd core). - Run the build:
python -m build. - The distributable files (
.whland.tar.gz) will be in thedist/directory.
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.
Quick Start with OpenAPI
from utcp_http.openapi_converter import OpenApiConverter
import aiohttp
# 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
# 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"
}
}]
})
Key Benefits
- ✅ Zero Infrastructure: No servers to deploy or maintain
- ✅ Direct API Calls: Native performance, no proxy overhead
- ✅ Automatic Conversion: OpenAPI schemas → UTCP tools
- ✅ Selective Authentication: Only protected endpoints get auth, public endpoints remain accessible
- ✅ Authentication Preserved: API keys, OAuth2, Basic auth supported
- ✅ Multi-format Support: JSON, YAML, OpenAPI 2.0/3.0
- ✅ Batch Processing: Convert multiple APIs simultaneously
Multiple Ingestion Methods
- Direct Converter:
OpenApiConverterclass for full control - Remote URLs: Fetch and convert specs from any URL
- Client Configuration: Include specs directly in UTCP config
- Batch Processing: Process multiple specs programmatically
- File-based: Convert local JSON/YAML specifications
📖 Complete OpenAPI Ingestion Guide - Detailed examples and advanced usage
