１关于UTCP
1.1 AI调用工具的发展–从MCP到UTCP
2024年AI最重要的事件是12月26日deepseek的发布，这个事件改变了AI世界的游戏规则，造成的影响可能数年甚至数十年后才能完全看的清楚。在deepseek发布前的不久，另外一个事件造成的影响也不容忽视，这就是11月Anthropic推出模型上下文协议（MCP），这是AI工具调用标准化的起点。
MCP的目的解决AI与外部工具集成的碎片化问题，它通过客户端-服务器架构（MCP客户端、主机、服务器）标准化工具调用流程，支持动态上下文联邦和分层安全设计，使AI能够安全访问实时数据源（如数据库、API），其优势在于兼容性（如OpenAI、百度智能云迅速接入）和企业级安全，但对代理层的依赖导致性能开销（平均延迟超800ms）较高，开发起来也比较复杂，应用于大并发的实时性场景明显尤其不足。
2025年7月，通用工具调用协议（UTCP） 提出“描述而非代理”的理念，直接砍掉中间服务器，允许AI代理通过JSON手册直连工具原生端点（HTTP/gRPC/CLI等）。UTCP的优势显著：
- 性能提升：延迟降低50%以上，无需代理转发的资源消耗；
- 开发简化：仅需JSON定义工具功能，无需编写适配代码；
- 协议兼容：支持MCP转换，并扩展至WebRTC、CLI等场景。
从MCP到UTCP，反映了AI工具调用从“统一接口”向“极致效率”的范式转变。MCP奠定了跨平台标准化的基础（如百度地图、支付宝的集成），而UTCP则通过“零包装”设计解决了MCP的固有瓶颈，更适合实时和高频调用场景（如自动驾驶、CLI工具）。未来，随着安全机制的进一步完善，UTCP有望成为AI代理与工具交互的新主流标准，推动AI生态向更高效、灵活的方向发展。
1.2 UTCP的设计原则
- 描述性，非代理性 (Descriptive, Not Proxied)：UTCP不代理流量。它通过一份机器可读的手册描述如何直接与工具的原生端点通信。
- 零包装 (Zero Wrapper Tax)：核心目标是消除开发、部署和维护包装器服务器的开销。
- 协议不可知论 (Protocol Agnostic)：该协议支持描述多种通信模式，包括HTTP、gRPC、WebSockets、Server-Sent Events (SSE) 甚至本地命令行接口 (CLI)。
- 自带安全机制 (Bring-Your-Own-Security)：工具保留其原有的认证、授权和安全模型（如API密钥、OAuth 2.0）。UTCP手册只是描述如何满足这些要求，而不是重新定义它们。
- AI原生 (AI-Native)：该协议从零开始设计，优先考虑AI代理的需求和能力。
1.3 UTCP的核心组件
UTCP的核心是一个结构化的手册（Manual），一个JSON文档，其中包含AI代理安全有效地使用工具所需的所有信息：
- 工具元数据：名称、版本和描述。
- 协议端点：工具原生接口（如OpenAPI规范URL、gRPC服务器地址、CLI命令路径）的详细信息。
- 操作（Operations）：工具可用功能的描述，包括输入参数和预期输出。
- 认证（Authentication）：描述工具所需的认证方法（例如，“在Authorization头中发送API密钥”）。
- 输入/输出模式：使用JSON Schema等标准定义操作期望的数据结构和类型。
1.4 UTCP的工作原理
1）发现：AI系统获取或发现工具的UTCP手册。
2）理解：AI解析手册以理解工具的功能、如何调用它们（协议、端点）以及需要哪些凭证（认证）。
3）调用：AI代理直接使用手册中指定的协议和认证信息向工具的原生端点发出请求。没有UTCP服务器中转流量。
4）交互：工具直接响应AI的请求，就像响应任何其他客户端一样。
2 智能deepseek查询系统python源码
介绍了UTCP的原理、特点和优势，下面是用python编写的一个简单的自然语言查询系统，每个部分都做了注释。
import mysql.connector
import json
import re
import requests # 使用requests库调用DeepSeek API
from datetime import datetime, date, time
from decimal import Decimal
from typing import Dict, List, Any, Optional
# 配置DeepSeek API密钥 - 请替换为您的实际密钥
DEEPSEEK_API_KEY = "sk-*********************************"
DEEPSEEK_API_URL = "https://api.deepseek.com/chat/completions"
# 自定义JSON编码器，处理各种非标准类型
class CustomJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (datetime, date, time)):
return obj.isoformat()
elif isinstance(obj, Decimal):
return float(obj)
elif hasattr(obj, '__dict__'):
return obj.__dict__
return super().default(obj)
# UTCP 手册定义，endoint部分填入自己的数据库连接信息，使用的是MySQL官方的airport示例数据库
UTCP_MANUAL = {
"tool_name": "MySQL_Database",
"description": "Direct connection to MySQL database using native connector",
"protocol": "mysql",
"endpoint": {
"host": "########",
"port": 3306,
"database": "airportdb",
"user": "root",
"password": "********"
},
"operations": {
"query": {
"type": "sql",
"description": "Execute SQL query and return results",
"parameters": {
"sql": {"type": "string", "description": "SQL statement to execute"}
}
}
},
"authentication": {
"type": "native",
"mechanism": "mysql_native_password"
},
"schema": {
"tables": [
{
"name": "users",
"columns": [
{"name": "id", "type": "INT", "description": "用户ID"},
{"name": "username", "type": "VARCHAR(50)", "description": "用户名"},
{"name": "email", "type": "VARCHAR(100)", "description": "电子邮件"},
{"name": "created_at", "type": "TIMESTAMP", "description": "创建时间"},
{"name": "last_login", "type": "TIMESTAMP", "description": "最后登录时间"},
{"name": "status", "type": "ENUM('active','inactive','suspended')", "description": "账户状态"},
{"name": "role", "type": "ENUM('user','admin','moderator')", "description": "用户角色"}
]
},
{
"name": "orders",
"columns": [
{"name": "order_id", "type": "INT", "description": "订单ID"},
{"name": "user_id", "type": "INT", "description": "用户ID"},
{"name": "product_name", "type": "VARCHAR(100)", "description": "产品名称"},
{"name": "amount", "type": "DECIMAL(10,2)", "description": "订单金额"},
{"name": "order_date", "type": "DATE", "description": "订单日期"}
]
}
]
}
}
class UTCP_MySQL_Connector:
"""UTCP 协议实现：直接连接MySQL数据库"""
def __init__(self, manual: dict):
self.manual = manual
self.connection = None
self._connect()
def _connect(self):
"""根据UTCP手册建立直接数据库连接"""
endpoint = self.manual["endpoint"]
try:
self.connection = mysql.connector.connect(
host=endpoint["host"],
port=endpoint["port"],
database=endpoint["database"],
user=endpoint["user"],
password=endpoint["password"],
pool_name="utcp_pool",
pool_size=5
)
print("✅ UTCP连接成功 | 原生MySQL连接已建立")
except mysql.connector.Error as err:
print(f"❌ UTCP连接失败: {err}")
raise ConnectionError(f"UTCP原生连接错误: {err}")
def execute_query(self, sql: str) -> Dict[str, Any]:
"""执行SQL查询并返回结果"""
cursor = self.connection.cursor(dictionary=True)
try:
cursor.execute(sql)
result = cursor.fetchall()
# 转换所有特殊类型为可序列化格式
def convert_value(value):
if isinstance(value, (datetime, date, time)):
return value.isoformat()
elif isinstance(value, Decimal):
return float(value)
elif isinstance(value, (bytes, bytearray)):
return value.decode('utf-8', errors='replace')
return value
converted_result = []
for row in result:
converted_row = {}
for key, value in row.items():
converted_row[key] = convert_value(value)
converted_result.append(converted_row)
return {
"status": "success",
"sql": sql,
"data": converted_result,
"row_count": len(converted_result),
"columns": [col[0] for col in cursor.description] if cursor.description else []
}
except mysql.connector.Error as err:
return {"status": "error", "message": str(err), "sql": sql}
finally:
cursor.close()
def close(self):
"""关闭原生连接"""
if self.connection:
self.connection.close()
print("🔌 UTCP连接已关闭")
class DeepSeekClient:
"""DeepSeek模型API客户端"""
def __init__(self, api_key: str, api_url: str = DEEPSEEK_API_URL):
self.api_key = api_key
self.api_url = api_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def generate_completion(self, messages: List[Dict], model: str = "deepseek-chat",
temperature: float = 0.7, max_tokens: int = 500) -> Optional[str]:
"""调用DeepSeek API生成文本"""
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = requests.post(self.api_url, headers=self.headers, json=payload)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content'].strip()
except requests.exceptions.RequestException as e:
print(f"❌ DeepSeek API请求失败: {str(e)}")
return None
except (KeyError, IndexError) as e:
print(f"❌ DeepSeek API响应解析失败: {str(e)}")
return None
class AIClient:
"""大模型客户端：智能数据库查询系统（DeepSeek集成版）"""
def __init__(self, utcp_connector: UTCP_MySQL_Connector, deepseek_client: DeepSeekClient):
self.connector = utcp_connector
self.deepseek = deepseek_client
self.history = []
def generate_sql(self, user_query: str) -> str:
"""
使用大模型生成SQL查询语句
参数:
user_query: 用户自然语言查询
返回:
SQL查询语句
"""
# 获取数据库模式信息
schema_info = json.dumps(UTCP_MANUAL["schema"], indent=2)
# 构建提示词
prompt = f"""
你是一个专业的SQL查询生成器。你的任务是根据用户的问题生成准确、安全的SQL查询语句。
只生成SQL语句，不要包含任何解释或其他文本。
数据库模式如下：
{schema_info}
用户查询：{user_query}
生成SQL查询时请注意：
1. 只使用SELECT语句，不要使用INSERT/UPDATE/DELETE等修改数据的语句
2. 确保表名和列名正确（注意大小写敏感）
3. 使用JOIN连接相关表
4. 包含必要的WHERE条件
5. 按需使用GROUP BY和聚合函数
6. 限制结果数量（使用LIMIT）
请直接输出SQL语句：
""".strip()
# 调用DeepSeek API生成SQL
messages = [
{"role": "system", "content": "你是一个SQL专家，只输出SQL语句，不包含任何其他文本。"},
{"role": "user", "content": prompt}
]
sql = self.deepseek.generate_completion(
messages,
model="deepseek-coder", # 使用DeepSeek的代码模型
temperature=0.2,
max_tokens=300
)
if not sql:
return None
# 确保只返回SQL语句（移除可能的代码块标记）
if sql.startswith("```sql"):
sql = re.sub(r'^```sql\s*|\s*```$', '', sql)
return sql
def explain_results(self, query: str, results: Dict[str, Any]) -> str:
"""
使用大模型解释查询结果
参数:
query: 原始用户查询
results: SQL查询结果
返回:
自然语言解释
"""
# 准备数据
data_str = json.dumps(results["data"][:5], indent=2, ensure_ascii=False) # 只显示前5行
if len(results["data"]) > 5:
data_str += f"\n...（共 {len(results['data'])} 条记录）"
prompt = f"""
你是一个数据分析助手，请根据用户的问题和查询结果，用自然语言解释结果。
回答要简洁明了，使用中文，并突出关键信息。
用户原始问题：{query}
执行的SQL查询：{results['sql']}
查询结果（JSON格式）：
{data_str}
请解释查询结果：
""".strip()
# 调用DeepSeek API生成解释
messages = [
{"role": "system", "content": "你是一个数据分析助手，请用简洁明了的中文解释查询结果。"},
{"role": "user", "content": prompt}
]
explanation = self.deepseek.generate_completion(
messages,
temperature=0.5,
max_tokens=600
)
return explanation or "抱歉，无法解释查询结果。"
def process_query(self, user_query: str) -> str:
"""
处理用户查询：生成SQL -> 执行查询 -> 解释结果
参数:
user_query: 用户自然语言查询
返回:
自然语言响应
"""
print(f"\n💬 用户查询: {user_query}")
# 步骤1: 生成SQL
print("🤖 正在生成SQL查询...")
sql = self.generate_sql(user_query)
if not sql:
return "抱歉，无法生成SQL查询。"
print(f"📝 生成的SQL: {sql}")
# 步骤2: 执行查询
print("🔍 正在执行数据库查询...")
results = self.connector.execute_query(sql)
if results["status"] != "success":
error_msg = f"数据库查询失败: {results.get('message', '未知错误')}"
print(f"❌ {error_msg}")
return error_msg
print(f"✅ 查询成功，返回 {results['row_count']} 条记录")
# 步骤3: 解释结果
print("💡 正在解释查询结果...")
explanation = self.explain_results(user_query, results)
# 保存历史记录
self.history.append({
"query": user_query,
"sql": sql,
"results": results,
"explanation": explanation,
"timestamp": datetime.now().isoformat()
})
return explanation
def get_history(self, limit: int = 5) -> List[Dict]:
"""获取最近的查询历史"""
return self.history[-limit:]
def close(self):
"""关闭数据库连接"""
self.connector.close()
def print_banner():
"""打印欢迎横幅"""
banner = """
**********************************************************************
* *
* 🧠 大模型客户端 - DeepSeek智能数据库查询系统 🧠 *
* *
* 功能： *
* 1. 理解自然语言查询 *
* 2. 生成SQL语句 *
* 3. 通过UTCP协议直接连接数据库 *
* 4. 以自然语言解释查询结果 *
* *
* 示例查询： *
* - "列出所有管理员用户" *
* - "显示每个用户的总订单金额" *
* - "查找最近一周的订单" *
* - "统计不同状态的用户数量" *
* *
* 使用模型：DeepSeek-Coder *
* *
**********************************************************************
""".strip()
print(banner)
def main():
"""主函数：启动大模型客户端"""
print_banner()
# 创建DeepSeek客户端
print("\n🧠 正在初始化DeepSeek模型客户端...")
deepseek_client = DeepSeekClient(api_key=DEEPSEEK_API_KEY)
# 创建UTCP连接器
print("\n🔗 正在初始化UTCP数据库连接...")
try:
connector = UTCP_MySQL_Connector(UTCP_MANUAL)
ai_client = AIClient(connector, deepseek_client)
except Exception as e:
print(f"❌ 初始化失败: {str(e)}")
return
print("\n✅ 系统已就绪，请输入您的查询（输入'exit'退出）")
# 主交互循环
while True:
try:
user_input = input("\n👤 您: ").strip()
if not user_input:
continue
if user_input.lower() in ["exit", "quit", "q"]:
print("\n👋 感谢使用，再见！")
break
if user_input.lower() == "history":
print("\n📜 查询历史:")
for i, item in enumerate(ai_client.get_history(5), 1):
print(f"{i}. [{item['timestamp']}] {item['query']}")
print(f" SQL: {item['sql']}")
continue
# 处理查询
response = ai_client.process_query(user_input)
print(f"\n🤖 助手: {response}")
except KeyboardInterrupt:
print("\n👋 操作已中断，输入'exit'退出")
except Exception as e:
print(f"\n❌ 发生错误: {str(e)}")
# 关闭连接
ai_client.close()
if __name__ == "__main__":
main()
3 运行智能数据库查询系统
3.1 准备运行环境
在运行这个系统之前，需要先安装python，也可以用虚拟环境运行，然后安装下面的依赖包
pip install requests mysql-connector-python
3.2 启动智能数据库查询系统
(langchainenv) C:\Users\lichx>python utcp-deepseek-mysql ********************************************************************** * * * 🧠 大模型客户端 - DeepSeek智能数据库查询系统 🧠 * * * * 功能： * * 1. 理解自然语言查询 * * 2. 生成SQL语句 * * 3. 通过UTCP协议直接连接数据库 * * 4. 以自然语言解释查询结果 * * * * 示例查询： * * - "列出所有管理员用户" * * - "显示每个用户的总订单金额" * * - "查找最近一周的订单" * * - "统计不同状态的用户数量" * * * * 使用模型：DeepSeek-Coder * * * ********************************************************************** 🧠 正在初始化DeepSeek模型客户端... 🔗 正在初始化UTCP数据库连接... ✅ UTCP连接成功 | 原生MySQL连接已建立 ✅ 系统已就绪，请输入您的查询（输入'exit'退出） #### 3.3 用自然语言查询数据库 👤 您: 列出所有管理员用户 💬 用户查询: 列出所有管理员用户 🤖 正在生成SQL查询... 📝 生成的SQL: SELECT * FROM users WHERE role = 'admin' LIMIT 100; 🔍 正在执行数据库查询... ✅ 查询成功，返回 2 条记录 💡 正在解释查询结果... 🤖 助手: 查询结果显示了2个管理员用户： 1. **admin_user** (ID:3) - 邮箱: admin@company.com - 创建时间: 2022年12月1日 - 最近登录: 2023年7月25日16:45 - 状态: 活跃 #### 4 小结 这个系统非常简单，离实际应用的距离还相当远，用来了解以下utcp协议还是不错的。 2. **system_operator** (ID:10) - 邮箱: sysop@operations.io - 创建时间: 2022年11月15日 - 最近登录: 2023年7月26日18:00 - 状态: 活跃 两个管理员账户当前均为活跃状态，且近期都有登录记录。
最后修改时间：2025-09-18 14:15:16
「喜欢这篇文章，您的关注和赞赏是给作者最好的鼓励」
关注作者
【版权声明】本文为墨天轮用户原创内容，转载时必须标注文章的来源（墨天轮），文章链接，文章作者等基本信息，否则作者和墨天轮有权追究责任。如果您发现墨天轮中有涉嫌抄袭或者侵权的内容，欢迎发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
