Your task is to analyze a given API specification (e.g., OpenAPI/Swagger, or plain text documentation) and convert it into a UTCPManual JSON object. This manual allows a UTCP client to understand and interact with the API's tools.
UTCPManual: The root JSON object that contains a list of all available tools from a provider. It has two main keys:versionandtools.Tool: A JSON object representing a single function or API endpoint. It describes what the tool does, what inputs it needs, what it returns, and how to call it.Provider: A JSON object inside aToolthat contains the specific connection details (e.g., HTTP URL, method, etc.).
Follow these steps to transform an API endpoint into a UTCP Tool.
Scan the API documentation and treat each unique API endpoint as a separate tool. For a REST API, an endpoint is a unique combination of an HTTP method and a URL path (e.g., GET /users/{id} is one tool, and POST /users is another).
For every endpoint you identify, you will create one JSON object that will be added to the tools array in the final UTCPManual.
This is the core of the task. Populate the fields of the Tool object as follows:
name: (String) Create a short, descriptive,snake_casename for the tool. Example:get_user_by_id.description: (String) Use the summary or description from the API documentation to explain what the tool does.tags: (Array of Strings) Add relevant keywords that can be used to search for this tool. These could be derived from the API's own tags or categories. Example:["users", "profile", "read"].average_response_size: (Integer, Optional) If the API documentation provides information on the typical size of the response payload in bytes, include it here. This is useful for performance considerations.inputs: (Object) A JSON Schema object describing all the parameters the API endpoint accepts (path, query, headers, and body).- Set
typeto"object". - In
properties, create a key for each parameter. The value should be an object defining itstype(e.g.,"string","number") anddescription. - In
required, create an array listing the names of all mandatory parameters.
- Set
outputs: (Object) A JSON Schema object describing the successful response from the API (e.g., the200 OKresponse body).- Set
typeto"object". - In
properties, map the fields of the JSON response body.
- Set
provider: (Object) This object contains the technical details needed to make the actual API call.provider_type: (String) Almost always"http"for web APIs.url: (String) The full URL of the endpoint. Use curly braces for path parameters, e.g.,https://api.example.com/users/{id}.http_method: (String) The HTTP method, e.g.,"GET","POST".content_type: (String) The request's content type, typically"application/json".path_fields: (Array of Strings) List the names of any parameters that are part of the URL path.header_fields: (Array of Strings) List the names of any parameters sent as request headers.body_field: (String) If the request has a JSON body, specify the name of the single input property that contains the body object.auth: (Object, Optional) If the API requires authentication, add this object. Theauth_typefield determines the authentication method (api_key,basic, oroauth2). Populate the other fields based on the API's security scheme. See theauth.pyreference below for the exact structure.
Once you have created a Tool object for every endpoint, assemble them into the final UTCPManual.
- Create the root JSON object.
- Set the
versionkey to"1.0". - Create a
toolskey with an array containing all theToolobjects you generated.
Below are the core Pydantic models that define the structure of a UTCPManual. Use these as the ground truth for the JSON structure you need to generate.
import inspect
from typing import Dict, Any, Optional, List, Literal, Union, get_type_hints
from pydantic import BaseModel, Field, TypeAdapter
from utcp.shared.provider import (
HttpProvider,
CliProvider,
WebSocketProvider,
GRPCProvider,
GraphQLProvider,
TCPProvider,
UDPProvider,
StreamableHttpProvider,
SSEProvider,
WebRTCProvider,
MCPProvider,
TextProvider,
)
class ToolInputOutputSchema(BaseModel):
type: str = Field(default="object")
properties: Dict[str, Any] = Field(default_factory=dict)
required: Optional[List[str]] = None
description: Optional[str] = None
title: Optional[str] = None
class Tool(BaseModel):
name: str
description: str = ""
inputs: ToolInputOutputSchema = Field(default_factory=ToolInputOutputSchema)
outputs: ToolInputOutputSchema = Field(default_factory=ToolInputOutputSchema)
tags: List[str] = []
average_response_size: Optional[int] = None
provider: Optional[Union[
HttpProvider,
CliProvider,
WebSocketProvider,
GRPCProvider,
GraphQLProvider,
TCPProvider,
UDPProvider,
StreamableHttpProvider,
SSEProvider,
WebRTCProvider,
MCPProvider,
TextProvider,
]] = Nonefrom typing import Literal, Optional, TypeAlias, Union
from pydantic import BaseModel, Field
class ApiKeyAuth(BaseModel):
"""Authentication using an API key.
The key can be provided directly or sourced from an environment variable.
"""
auth_type: Literal["api_key"] = "api_key"
api_key: str = Field(..., description="The API key for authentication.")
var_name: str = Field(
..., description="The name of the variable containing the API key."
)
class BasicAuth(BaseModel):
"""Authentication using a username and password."""
auth_type: Literal["basic"] = "basic"
username: str = Field(..., description="The username for basic authentication.")
password: str = Field(..., description="The password for basic authentication.")
class OAuth2Auth(BaseModel):
"""Authentication using OAuth2."""
auth_type: Literal["oauth2"] = "oauth2"
token_url: str = Field(..., description="The URL to fetch the OAuth2 token from.")
client_id: str = Field(..., description="The OAuth2 client ID.")
client_secret: str = Field(..., description="The OAuth2 client secret.")
scope: Optional[str] = Field(None, description="The OAuth2 scope.")
Auth: TypeAlias = Union[ApiKeyAuth, BasicAuth, OAuth2Auth]from typing import Dict, Any, Optional, List, Literal, TypeAlias, Union
from pydantic import BaseModel, Field
from utcp.shared.auth import (
Auth,
ApiKeyAuth,
BasicAuth,
OAuth2Auth,
)
ProviderType: TypeAlias = Literal[
'http', # RESTful HTTP/HTTPS API
'sse', # Server-Sent Events
'http_stream', # HTTP Chunked Transfer Encoding
'cli', # Command Line Interface
'websocket', # WebSocket bidirectional connection
'grpc', # gRPC (Google Remote Procedure Call)
'graphql', # GraphQL query language
'tcp', # Raw TCP socket
'udp', # User Datagram Protocol
'webrtc', # Web Real-Time Communication
'mcp', # Model Context Protocol
'text', # Text file provider
]
class Provider(BaseModel):
name: str
provider_type: ProviderType
startup_command: Optional[List[str]] = None # For launching the provider if needed
class HttpProvider(Provider):
"""Options specific to HTTP tools"""
provider_type: Literal["http"] = "http"
http_method: Literal["GET", "POST", "PUT", "DELETE", "PATCH"] = "GET"
url: str
content_type: str = "application/json"
auth: Optional[Auth] = None
headers: Optional[Dict[str, str]] = None
body_field: Optional[str] = Field(default=None, description="The name of the single input field to be sent as the request body.")
header_fields: Optional[List[str]] = Field(default=None, description="List of input fields to be sent as request headers.")
class SSEProvider(Provider):
"""Options specific to Server-Sent Events tools"""
provider_type: Literal["sse"] = "sse"
url: str
event_type: Optional[str] = None
reconnect: bool = True
retry_timeout: int = 30000 # Retry timeout in milliseconds if disconnected
auth: Optional[Auth] = None
headers: Optional[Dict[str, str]] = None
body_field: Optional[str] = Field(default=None, description="The name of the single input field to be sent as the request body.")
header_fields: Optional[List[str]] = Field(default=None, description="List of input fields to be sent as request headers for the initial connection.")
class StreamableHttpProvider(Provider):
"""Options specific to HTTP Chunked Transfer Encoding (HTTP streaming) tools"""
provider_type: Literal["http_stream"] = "http_stream"
url: str
http_method: Literal["GET", "POST"] = "GET"
content_type: str = "application/octet-stream"
chunk_size: int = 4096 # Size of chunks in bytes
timeout: int = 60000 # Timeout in milliseconds
headers: Optional[Dict[str, str]] = None
auth: Optional[Auth] = None
body_field: Optional[str] = Field(default=None, description="The name of the single input field to be sent as the request body.")
header_fields: Optional[List[str]] = Field(default=None, description="List of input fields to be sent as request headers.")
class CliProvider(Provider):
"""Options specific to CLI tools"""
provider_type: Literal["cli"] = "cli"
command_name: str
env_vars: Optional[Dict[str, str]] = Field(default=None, description="Environment variables to set when executing the command")
working_dir: Optional[str] = Field(default=None, description="Working directory for command execution")
auth: None = None
class WebSocketProvider(Provider):
"""Options specific to WebSocket tools"""
provider_type: Literal["websocket"] = "websocket"
url: str
protocol: Optional[str] = None
keep_alive: bool = True
auth: Optional[Auth] = None
headers: Optional[Dict[str, str]] = None
header_fields: Optional[List[str]] = Field(default=None, description="List of input fields to be sent as request headers for the initial connection.")
class GRPCProvider(Provider):
"""Options specific to gRPC tools"""
provider_type: Literal["grpc"] = "grpc"
host: str
port: int
service_name: str
method_name: str
use_ssl: bool = False
auth: Optional[Auth] = None
class GraphQLProvider(Provider):
"""Options specific to GraphQL tools"""
provider_type: Literal["graphql"] = "graphql"
url: str
operation_type: Literal["query", "mutation", "subscription"] = "query"
operation_name: Optional[str] = None
auth: Optional[Auth] = None
headers: Optional[Dict[str, str]] = None
header_fields: Optional[List[str]] = Field(default=None, description="List of input fields to be sent as request headers for the initial connection.")
class TCPProvider(Provider):
"""Options specific to raw TCP socket tools"""
provider_type: Literal["tcp"] = "tcp"
host: str
port: int
timeout: int = 30000
auth: None = None
class UDPProvider(Provider):
"""Options specific to UDP socket tools"""
provider_type: Literal["udp"] = "udp"
host: str
port: int
timeout: int = 30000
auth: None = None
class WebRTCProvider(Provider):
"""Options specific to WebRTC tools"""
provider_type: Literal["webrtc"] = "webrtc"
signaling_server: str
peer_id: str
data_channel_name: str = "tools"
auth: None = None
class McpStdioServer(BaseModel):
"""Configuration for an MCP server connected via stdio."""
transport: Literal["stdio"] = "stdio"
command: str
args: Optional[List[str]] = []
env: Optional[Dict[str, str]] = {}
class McpHttpServer(BaseModel):
"""Configuration for an MCP server connected via streamable HTTP."""
transport: Literal["http"] = "http"
url: str
McpServer: TypeAlias = Union[McpStdioServer, McpHttpServer]
class McpConfig(BaseModel):
mcpServers: Dict[str, McpServer]
class MCPProvider(Provider):
"""Options specific to MCP tools, supporting both stdio and HTTP transports."""
provider_type: Literal["mcp"] = "mcp"
config: McpConfig
auth: Optional[OAuth2Auth] = None
class TextProvider(Provider):
"""Options specific to text file-based tools.
This provider reads tool definitions from a local text file. This is useful
when the tool call is included in the startup command, but the result of the
tool call produces a file at a static location that can be read from. It can
also be used as a UTCP tool provider to specify tools that should be used
from different other providers.
"""
provider_type: Literal["text"] = "text"
file_path: str = Field(..., description="The path to the file containing the tool definitions.")
auth: None = NoneAPI Specification Snippet:
Endpoint: GET /v1/weather
Description: Retrieves the current weather for a specific city.
Query Parameters:
- `city` (string, required): The name of the city (e.g., "London").
- `units` (string, optional): The temperature units. Can be 'metric' or 'imperial'. Defaults to 'metric'.
Response (200 OK):
{
"temperature": 15,
"conditions": "Cloudy",
"humidity": 82
}
Generated UTCPManual:
{
"version": "1.0",
"tools": [
{
"name": "get_weather",
"description": "Retrieves the current weather for a specific city.",
"inputs": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city (e.g., \"London\")."
},
"units": {
"type": "string",
"description": "The temperature units. Can be 'metric' or 'imperial'."
}
},
"required": [
"city"
]
},
"outputs": {
"type": "object",
"properties": {
"temperature": {
"type": "number"
},
"conditions": {
"type": "string"
},
"humidity": {
"type": "number"
}
}
},
"tool_provider": {
"name": "weather_service",
"provider_type": "http",
"url": "https://api.example.com/v1/weather",
"http_method": "GET",
"content_type": "application/json"
}
}
]
}