分享开发技术
0%

UTCP的介绍和使用

发表于
本文字数： 12k 阅读时长 ≈ 11 分钟

UTCP是一个工具调用协议，全称是 Universal Tool Calling Protocol，相比于前段时间火热的MCP（Model Context Protocol）协议，它提供了更简单灵活的工具调用方式，它使 AI 代理和应用程序能够使用其本机协议直接发现和调用工具而无需封装成服务器。

UTCP 主要由四个部分组成：

1
2
3
4
Manuals: Manuals是标准的工具提供程序的描述格式，它包含了tool的定义信息
Tools: Tools是可以被调用的一系列功能
Call Templates: Call Templates是一套通信配置，它定义了如何访问tools。具体来说，Call Templates将tool名称和tool参数映射到一个通信协议的api请求中，通信协议可以是http、websockets、graphql等
UtcpClient: UtcpClient是使用Call Templates调用tool的客户端

UTCP的文档地址：https://www.utcp.io/

UTCP的安装

1
2
3
4
5
# 安装core + http插件
pip install utcp utcp-http

# 安装其它所需的插件
pip install utcp-cli utcp-mcp utcp-text

UTCP的基本使用

使用http协议定义Manual

HTTP 协议插件使 UTCP 能够通过 HTTP/HTTPS 请求调用工具，是 REST API、Webhook 和 Web 服务最常用的协议。

定义一个不使用 UTCP 依赖的 Manual：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import uvicorn
from fastapi import FastAPI

app = FastAPI()


SERVER_IP = '127.0.0.1'
SERVER_PORT = 8012
BASE_URL = f"http://{SERVER_IP}:{SERVER_PORT}"


@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",
                "inputs": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "City name"
                        }
                    },
                    "required": ["location"]
                },
                "outputs": {
                    "type": "object",
                    "properties": {
                        "temperature": {"type": "number"},
                        "conditions": {"type": "string"}
                    }
                },
                "tool_call_template": {
                    "call_template_type": "http",
                    "url": f"{BASE_URL}/api/weather",
                    "http_method": "GET"
                }
            },
            {
                "name": "book_ticket",
                "description": "根据出发地、目的地以及日期预订机票",
                "inputs": {
                    "type": "object",
                    "properties": {
                        "source": {
                            "type": "string",
                            "description": "出发地"
                        },
                        "dest": {
                            "type": "string",
                            "description": "目的地"
                        },
                        "date": {
                            "type": "string",
                            "description": "日期，格式为：2025-08-31"
                        }
                    },
                    "required": ["source", "dest", "data"]
                },
                "outputs": {
                    "type": "object",
                    "properties": {
                        "status": {"type": "string"}
                    }
                },
                "tool_call_template": {
                    "call_template_type": "http",
                    "url": f"{BASE_URL}/api/book_ticket",
                    "http_method": "POST"
                }
            }
        ]
    }


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


@app.post("/api/book_ticket")
def get_weather(source: str, dest: str, date: str):
    return {"status": "success"}


if __name__ == "__main__":
    uvicorn.run(
        app,
        host=SERVER_IP,
        port=SERVER_PORT
    )

上面的例子中定义了一个Mauual，即 /utcp 接口中返回的一个字典格式的数据，该 manual 作为一个 http 接口的响应数据暴露出去，供其它 UtcpClient 客户端调用。这里不需要安装其它 UTCP 相关的依赖。

该 manual 中定义了若干个 tool，每个 tool 中定义了名称、输入、输出以及 Call Templates。

其中 Call Templates 中定义了如何访问该 tool，包括：访问协议类型、URL、请求方法等。Call Templates 中指定的 URL 就是我们上面定义的 http 接口。

定义 UtcpClient：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import asyncio
from utcp.utcp_client import UtcpClient


async def main():
    client = await UtcpClient.create(config={
        "manual_call_templates": [
            {
                "name": "client_service",
                "call_template_type": "http",
                "http_method": "GET",
                "url": "http://127.0.0.1:8012/utcp",
                "content_type": "application/json"
            }
        ]
    })

    # 查询所有工具
    tools = await client.search_tools("")
    print(f"tools: {tools}")
    
    result = await client.call_tool(
        "client_service.get_weather",
        tool_args={"location": "Wuhan"}
    )
    print(f"Weather: {result['temperature']}°C, {result['conditions']}")

    result = await client.call_tool(
        "client_service.book_ticket",
        tool_args={"source": "Wuhan", "dest": "Beijing", "date": "2025-08-31"}
    )
    print(f"预定结果: {result['status']}")


if __name__ == "__main__":
    asyncio.run(main())

上面的代码中创建了一个 UtcpClient，UtcpClient中定义了 manual_call_templates，manual_call_templates 中指定了我们要访问的 Mauual 信息，创建了 UtcpClient 后就可以调用 Manual 中定义的 tool 了。

上面就是一个简单的使用 HTTP 协议的 tool 调用例子，通过上面的例子可以看出，MCP协议与UTCP 协议的一点区别是：MCP需要编写 server 端代码，UTCP需要编写 Manual，但是Manual相对来说简单一些，而且更加灵活，通过 http 接口就可以暴露出去。

还可以通过 UTCP 包的装饰器来定义 Manual：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import uvicorn
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()


SERVER_IP = '127.0.0.1'
SERVER_PORT = 8012
BASE_URL = f"http://{SERVER_IP}:{SERVER_PORT}"


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


# 使用装饰器修饰接口，接口就变成了utcp tool
@utcp_tool(
    tool_call_template=HttpCallTemplate(
        name='get_weather',
        url=f'{BASE_URL}/api/weather',
        http_methpd='GET'
    ),
    tags=['weather']
)
@app.get("/api/weather")
async def get_weather(location: str):
    """根据城市名成获取城市的天气信息
    :param location: 城市名称
    """
    return {"temperature": 22.5, "conditions": "Sunny"}


@utcp_tool(
    tool_call_template=HttpCallTemplate(
        name='book_ticket',
        url=f'{BASE_URL}/api/book_ticket',
        http_method='POST'
    ),
    tags=['ticket']
)
@app.post("/api/book_ticket")
async def book_ticket(source: str, dest: str, date: str):
    """根据出发地、目的地、日期预定机票
    :param source: 出发地
    :param dest: 目的地
    :param date: 日期
    """
    return {"status": "success"}


if __name__ == "__main__":
    uvicorn.run(
        app,
        host=SERVER_IP,
        port=SERVER_PORT
    )

使用streamable_http协议定义Manual

Streamable HTTP 协议使 UTCP 能够通过分块流式传输大型 HTTP 响应。这对于返回大型数据集、文件或无法装入单个响应的结果的工具而言非常理想。

下面是一个 streamable_http manual 的例子，tool_call_template 的配置内容如下：

1
2
3
4
5
6
7
{
	"call_template_type": "streamable_http",     // 协议类型
	"content_type": "application/octet-stream",  // 返回的内容类型
	"url": f "{base_url}/api/weather",           // http接口地址
	"chunk_size": 10,                            // 响应流中每个数据块的大小
	"http_method": "GET"                         // 请求方法
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import uvicorn
from fastapi import FastAPI
from fastapi.responses import StreamingResponse


app = FastAPI()


SERVER_IP = '127.0.0.1'
SERVER_PORT = 8012


@app.get("/utcp")
async def utcp_discovery():
    base_url = f"http://{SERVER_IP}:{SERVER_PORT}"
    return {
        "manual_version": "1.0.0",
        "utcp_version": "1.0.1",
        "tools": [
            {
                "name": "get_weather",
                "description": "Get current weather for a location",
                "inputs": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "City name"
                        }
                    },
                    "required": ["location"]
                },
                "outputs": {
                    "type": "object",
                    "properties": {
                        "temperature": {"type": "number"},
                        "conditions": {"type": "string"},
                        "message": {"type": "string"},
                    }
                },
                "tool_call_template": {
                    "call_template_type": "streamable_http",
                    "content_type": "application/octet-stream",
                    "url": f"{base_url}/api/weather",
                    "chunk_size": 10,
                    "http_method": "GET"
                }
            }
        ]
    }


@app.get("/api/weather")
async def get_weather(location: str):
    async def generate():
        data = ['{"temperature": 22.5, ', '"conditions": "Sunny", ', '"message": "天气很好，很适合出去游玩露营"}']
        for d in data:
            yield d

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )


if __name__ == "__main__":
    uvicorn.run(
        app,
        host=SERVER_IP,
        port=SERVER_PORT,
        log_level="info"
    )

UtcpClient 的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import asyncio
from utcp.utcp_client import UtcpClient


async def main():
    client = await UtcpClient.create(config={
        "manual_call_templates": [
            {
                "name": "weather_service",
                "call_template_type": "http",
                "http_method": "GET",
                "url": "http://127.0.0.1:8012/utcp",
                "content_type": "application/json"
            }
        ]
    })

    # 流式输出结果
    result = client.call_tool_streaming(
        "weather_service.get_weather",
        tool_args={"location": "Wuhan"}
    )

    full_content = ''
    async for r in result:
        content = r.decode('utf-8', errors='ignore')
        full_content += content
    print(f"full content: {full_content}")


if __name__ == "__main__":
    asyncio.run(main())

Manual中配置权限校验

Manual 中需要增加 auth 的配置：

1
2
3
4
5
6
{
	"auth_type": "api_key",
	"api_key": "Bearer ${API_KEY}",
	"var_name": "Authorization",
	"location": "header"
}

完整代码为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from typing import Optional

import uvicorn
from fastapi import FastAPI, Depends, HTTPException, status, Header


app = FastAPI()


SERVER_IP = '127.0.0.1'
SERVER_PORT = 8012
BASE_URL = f"http://{SERVER_IP}:{SERVER_PORT}"


async def validate_access_token(authorization: Optional[str] = Header(None)):
    """校验access_token
    """
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing or invalid Authorization header",
            headers={"WWW-Authenticate": "Bearer"},
        )

    token = authorization.replace("Bearer ", "").strip()

    if token != "123456":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return token


@app.get("/utcp")
async 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",
                "inputs": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "City name"
                        }
                    },
                    "required": ["location"]
                },
                "outputs": {
                    "type": "object",
                    "properties": {
                        "temperature": {"type": "number"},
                        "conditions": {"type": "string"},
                        "message": {"type": "string"},
                    }
                },
                "tool_call_template": {
                    "call_template_type": "http",
                    "url": f"{BASE_URL}/api/weather",
                    "http_method": "GET",
                    "content_type": "application/json",
                    "auth": {
                        "auth_type": "api_key",
                        "api_key": "Bearer ${API_KEY}",
                        "var_name": "Authorization",
                        "location": "header"
                      }
                }
            }
        ]
    }


@app.get("/api/weather")
async def get_weather(location: str, access_token: str = Depends(validate_access_token)):
    """根据城市名成获取城市的天气信息
        :param location: 城市名称
        :param access_token: token
        """
    return {"temperature": 22.5, "conditions": "Sunny"}


if __name__ == "__main__":
    uvicorn.run(
        app,
        host=SERVER_IP,
        port=SERVER_PORT,
        log_level="info"
    )

UtcpClient 的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import asyncio
from utcp.utcp_client import UtcpClient


async def main():
    client = await UtcpClient.create(config={
        "variables": {
            # 配置环境变量，这里的环境变量名称是由call template的name加上变量名构成的
            "client__service_API_KEY": "123456"
        },
        "manual_call_templates": [
            {
                "name": "client_service",
                "call_template_type": "http",
                "http_method": "GET",
                "url": "http://127.0.0.1:8012/utcp",
                "content_type": "application/json"
            }
        ]
    })

    # 查询所有工具
    tools = await client.search_tools("")
    print(f"tools: {tools}")

    result = await client.call_tool(
        "client_service.get_weather",
        tool_args={"location": "Wuhan"}
    )
    print(f"Weather: {result['temperature']}°C, {result['conditions']}")


if __name__ == "__main__":
    asyncio.run(main())

也可以通过 UTCP 的装饰器添加含有权限校验的tool，使用 ApiKeyAuth 添加基于 api_key 的权限校验，接口如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from utcp.data.auth_implementations import ApiKeyAuth

@utcp_tool(
    tool_call_template=HttpCallTemplate(
        name='get_weather',
        url=f"{BASE_URL}/api/weather",
        http_method='GET',
        auth=ApiKeyAuth(api_key="Bearer ${API_KEY}", var_name="Authorization", location="header"),
    )
)
@app.get("/api/weather")
async def get_weather(location: str, access_token: str = Depends(validate_access_token)):
    """根据城市名成获取城市的天气信息
        :param location: 城市名称
        :param access_token: token
        """
    return {"temperature": 22.5, "conditions": "Sunny"}

还可以添加基于 BasicAuth 和 OAuth2Auth 的权限校验。

基于特定通信协议调用工具

UTCP 中还提供了若干个类用来调用工具，这些类都是基于UTCP客户端的不同通信协议的实现，包括：HttpCommunicationProtocol、StreamableHttpCommunicationProtocol、SseCommunicationProtocol。

下面是一个 UTCP 客户端的例子，使用了 HttpCommunicationProtocol 来调用工具：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import asyncio

from utcp_http.http_communication_protocol import HttpCommunicationProtocol
from utcp_http.http_call_template import HttpCallTemplate
from utcp.data.auth_implementations import ApiKeyAuth


async def http_call_temp_api_key(name, tool_url, api_key, var_name, location='header', http_method='GET'):
    """创建一个http call template
    """
    return HttpCallTemplate(
        name=name,
        url=tool_url,
        http_method=http_method,
        auth=ApiKeyAuth(
            api_key=api_key,
            var_name=var_name,
            location=location,
        )
    )


async def main():
    http_call_template = await http_call_temp_api_key(
        name='get_weather',
        tool_url='http://127.0.0.1:8012/api/weather',
        api_key='Bearer 123456',
        var_name='Authorization'
    )

    http_transport = HttpCommunicationProtocol()
    result = await http_transport.call_tool(None, 'get_weather',
                                            {"location": "Wuhan"}, 
                                            http_call_template)
    print(f"result: {result}")


if __name__ == "__main__":
    asyncio.run(main())