官方 Python 实现的 UTCP 1.0.1 正式发布

通用工具调用协议(UTCP)是一种现代、灵活且可扩展的标准，用于定义和与各种通信协议中的工具进行交互。UTCP 1.0.0 引入了基于插件架构的模块化核心，使其更具可扩展性、可测试性，并更易于打包。

与其他协议相比，UTCP 特别强调以下几点：

• 可扩展性：UTCP 设计上能够处理大量工具和提供商，同时不牺牲性能。
• 可扩展性：可插拔架构使开发人员能够轻松添加新的通信协议、工具存储机制和搜索策略，而无需修改核心库。
• 互操作性：随着协议插件生态系统的不断扩展(包括 HTTP、SSE、CLI 等)，UTCP 可与几乎任何现有服务或基础设施集成。
• 易用性： 该协议基于简单、定义明确的 Pydantic 模型构建，使开发人员能够轻松实现和使用。

1.0.0 版本的新架构

UTCP 已重构为一个核心库和一组可选插件。

核心包(utcp)

utcp 包提供核心组件和接口：

• 数据模型：用于 Tool、CallTemplate、UtcpManual 和 Auth 的 Pydantic 模型。
• 可插拔接口：
• CommunicationProtocol：定义协议特定通信的契约(例如 HTTP、CLI)。
• ConcurrentToolRepository：用于存储和检索工具的线程安全访问接口。
  • ToolSearchStrategy：用于实现工具搜索算法的接口。
• VariableSubstitutor：处理配置中的变量替换。
• ToolPostProcessor：允许在返回工具结果前对其进行修改。
• 默认实现：
• UtcpClient：与 UTCP 生态系统交互的主要客户端。
  • InMemToolRepository：一个内存工具库，支持异步读写锁。
  • TagAndDescriptionWordMatchStrategy：一种改进的搜索策略，支持基于标签和描述关键词的匹配。

协议插件

通信协议现已作为独立的可安装包。这使核心保持精简，并允许用户仅安装所需的协议。

• utcp-http：支持 HTTP、SSE 和可流式传输的 HTTP，以及 OpenAPI 转换器。
• utcp-cli：用于封装本地命令行工具。
• utcp-mcp：用于与模型上下文协议(MCP)的互操作性。
• utcp-text：用于读取文本文件。
• utcp-socket：TCP 和 UDP 协议的框架。(正在开发中，需要更新)
• utcp-gql：GraphQL 的框架。(正在开发中，需要更新)

安装

安装核心库及所需协议插件。

# Install the core client and the HTTP plugin
pip install utcp utcp-http

# Install the CLI plugin as well
pip install utcp-cli

开发时，可从克隆的仓库以可编辑模式安装包：

# 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

从 0.x 到 1.0.0 的迁移指南

版本 1.0.0 引入了多个兼容性更改。请按照以下步骤迁移您的项目。

1. 更新依赖项：安装新的 utcp 核心包以及您使用的特定协议插件(例如 utcp-http、utcp-cli)。
2. 配置：

• 配置对象：UtcpClient 通过 UtcpClientConfig 对象、字典或指向包含配置的 JSON 文件的路径进行初始化。
  • 手动调用模板：providers_file_path 选项已被移除。现在，您需要直接在 UtcpClientConfig 中提供 manual_call_templates 的列表，而非文件路径。
• 术语：术语 provider 已被替换为 call_template，而 provider_type 现为 call_template_type。
  • 可流式 HTTP：call_template_type 中的 http_stream 已重命名为 streamable_http。

1. 更新导入：将导入语句修改为反映新的模块化结构。例如，from utcp.client.transport_interfaces.http_transport import HttpProvider 改为 from utcp_http.http_call_template import HttpCallTemplate。
2. 工具搜索：如果您之前使用的是默认搜索策略，新的策略为 TagAndDescriptionWordMatchStrategy。这是新的默认策略，除非您之前实现了自定义策略，否则无需进行任何更改。
3. 工具命名：工具名称现在以 manual_name.tool_name 的形式进行命名空间划分。客户端会自动处理此操作。6. 变量替换命名空间：在不同 call_templates 中替换的变量，首先会使用手册名称并重复 _ 进行命名空间划分。例如，来自手册 manual_1 的工具调用模板中的 API_KEY 键将转换为 manual__1_API_KEY。

使用示例

1. 使用 UTCP 客户端

config.json(可选)

您可以通过 JSON 文件定义完整的客户端配置。所有字段均为可选。

{
  "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. 提供 UTCP 手册

UTCPManual 描述您提供的工具。关键变化是将 tool_provider 替换为 call_template。

server.py

UTCP装饰器版本：

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"}

无UTCP依赖的服务器版本：

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.1",
        "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"}
                    }
                },
                "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. 完整示例

您可以在 示例仓库 中找到完整示例。

协议规范

UtcpManual 和 Tool 模型

Tool 中的 tool_provider 对象已被 call_template 取代。

{
  "manual_version": "string",
  "utcp_version": "string",
  "tools": [
    {
      "name": "string",
      "description": "string",
      "inputs": { ... },
      "outputs": { ... },
      "tags": ["string"],
      "call_template": {
        "call_template_type": "http",
        "url": "https://...",
        "http_method": "GET"
      }
    }
  ]
}

调用模板配置示例

各协议的配置示例。请记得将 provider_type 替换为 call_template_type。

HTTP 调用模板

{
  "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"
  "auth": { // Optional, example using ApiKeyAuth for a Bearer token. The client must prepend "Bearer " to the token.
    "auth_type": "api_key",
    "api_key": "Bearer $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(服务器发送事件)调用模板

{
  "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
}

可流式 HTTP 调用模板

注意名称已从 http_stream 更改为 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 调用模板

{
  "name": "my_cli_tool",
  "call_template_type": "cli", // Required
  "command_name": "my-command --utcp", // Required
  "env_vars": { // Optional
    "MY_VAR": "my_value"
  },
  "working_dir": "/path/to/working/directory", // Optional
  "auth": null // Optional (always null for CLI)
}

文本调用模板

{
  "name": "my_text_manual",
  "call_template_type": "text", // Required
  "file_path": "./manuals/my_manual.json", // Required
  "auth": null // Optional (always null for Text)
}

MCP(模型上下文协议)调用模板

{
  "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
  }
}

测试

测试结构已更新以反映新的核心/插件分离。

运行测试

要运行核心库和所有插件的所有测试：

# Ensure you have installed all dev dependencies
python -m pytest

要运行特定包(例如核心库)的测试：

python -m pytest core/tests/

要运行特定插件(例如 HTTP)的测试：

python -m pytest plugins/communication_protocols/http/tests/ -v

要运行带覆盖率的测试：

python -m pytest --cov=utcp --cov-report=xml

构建

构建过程现在涉及根据需要分别构建每个包(core 和 plugins)，尽管它们独立发布到 PyPI。

1. 创建并激活虚拟环境。
2. 安装构建依赖项：pip install build。
3. 导航至包目录(例如：cd core)。
4. 运行构建：python -m build。
5. 可分发文件(.whl 和 .tar.gz)将位于 dist/ 目录中。

你也许感兴趣的：

• 人工智能对 Python 的严重偏爱：大语言模型偏爱一种语言的问题
• Python 性能神话与传说
• Python 如何从一门编程语言发展成为一个社区
• Lamport 的拜占庭将军算法的 Python 实现
• 30个简单Python项目的轻松解决与解析
• Python 3.14 中的 Zstandard 压缩：为何这对开发者至关重要
• 关于 Python JIT 的后续进展
• 如何改进Python打包，或者为什么14个工具至少有12个是多余的
• 我正在转向 Python 并且真的喜欢它
• Python 正在逐步移除 GIL，这对 Python 开发者意味着什么