Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-23 23:53:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

update:智控台，完成mcp拍照识图

Browse Source
This commit is contained in:
hrz
2025-06-01 13:34:32 +08:00
parent c4f2411fee
commit ce776f210c
19 changed files with 267 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/app.py
+5 -1
View File
@@ -1,4 +1,5 @@
import sys
import uuid
import signal
import asyncio
from aioconsole import ainput
@@ -45,13 +46,16 @@ async def main():
check_ffmpeg_installed()
config = load_config()
# 生成随机密钥并添加到配置中
config["server"]["auth_key"] = str(uuid.uuid4().hex)
# 添加 stdin 监控任务
stdin_task = asyncio.create_task(monitor_stdin())
# 启动 WebSocket 服务器
ws_server = WebSocketServer(config)
ws_task = asyncio.create_task(ws_server.start())
# 启动 Simple OTA 服务器
# 启动 Simple http 服务器
ota_server = SimpleHttpServer(config)
ota_task = asyncio.create_task(ota_server.start())
main/xiaozhi-server/config/config_loader.py
+4
View File
@@ -59,10 +59,14 @@ def get_config_from_api(config):
"url": config["manager-api"].get("url", ""),
"secret": config["manager-api"].get("secret", ""),
}
# server的配置以本地为准
if config.get("server"):
config_data["server"] = {
"ip": config["server"].get("ip", ""),
"port": config["server"].get("port", ""),
"http_port": config["server"].get("http_port", ""),
"vision_explain": config["server"].get("vision_explain", ""),
"auth_key": config["server"].get("auth_key", ""),
}
return config_data
main/xiaozhi-server/config_from_api.yaml
+9
View File
@@ -8,6 +8,15 @@
server:
ip: 0.0.0.0
port: 8000
# http服务的端口，用于视觉分析接口
http_port: 8003
# 视觉分析接口地址
# 向设备发送的视觉分析的接口地址
# 如果按下面默认的写法，系统会自动生成视觉识别地址，并输出在启动日志里，这个地址你可以直接用浏览器访问确认一下
# 当你使用docker部署或使用公网部署(使用ssl、域名)时，不一定准确
# 所以如果你使用docker部署时，将vision_explain设置成局域网地址
# 如果你使用公网部署时，将vision_explain设置成公网地址
vision_explain: http://你的ip或者域名:端口号/mcp/vision/explain
manager-api:
# 你的manager-api的地址，最好使用局域网ip
url: http://127.0.0.1:8002/xiaozhi
main/xiaozhi-server/core/api/vision_handler.py
+67 -7
View File
@@ -1,9 +1,13 @@
import json
import copy
from aiohttp import web
from config.logger import setup_logging
from core.utils.util import get_vision_url, is_valid_image_file
from core.utils.vllm import create_instance
from config.config_loader import get_private_config_from_api
from core.utils.auth import AuthToken
import base64
from typing import Tuple, Optional
TAG = __name__
@@ -15,20 +19,45 @@ class VisionHandler:
def __init__(self, config: dict):
self.config = config
self.logger = setup_logging()
# 初始化认证工具
self.auth = AuthToken(config["server"]["auth_key"])
def _create_error_response(self, message: str) -> dict:
"""创建统一的错误响应格式"""
return {"success": False, "message": message}
def _verify_auth_token(self, request) -> Tuple[bool, Optional[str]]:
"""验证认证token"""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return False, None
token = auth_header[7:] # 移除"Bearer "前缀
return self.auth.verify_token(token)
async def handle_post(self, request):
"""处理 MCP Vision POST 请求"""
try:
# 验证token
is_valid, token_device_id = self._verify_auth_token(request)
if not is_valid:
return web.Response(
text=json.dumps(
self._create_error_response("无效的认证token或token已过期")
),
content_type="application/json",
status=401,
)
# 获取请求头信息
device_id = request.headers.get("Device-Id", "")
client_id = request.headers.get("Client-Id", "")
self.logger.bind(tag=TAG).debug(f"Device-Id: {device_id}")
self.logger.bind(tag=TAG).debug(f"Client-Id: {client_id}")
if device_id != token_device_id:
return web.Response(
text=json.dumps(self._create_error_response("设备ID与token不匹配")),
content_type="application/json",
status=401,
)
# 解析multipart/form-data请求
reader = await request.multipart()
@@ -64,9 +93,34 @@ class VisionHandler:
# 将图片转换为base64编码
image_base64 = base64.b64encode(image_data).decode("utf-8")
vllm_config = self.config["VLLM"]["ChatGLMVLLM"]
provider = create_instance("openai", vllm_config)
response = provider.response(question, image_base64)
# 如果开启了智控台，则从智控台获取模型配置
current_config = copy.deepcopy(self.config)
read_config_from_api = current_config.get("read_config_from_api", False)
if read_config_from_api:
current_config = get_private_config_from_api(
current_config,
device_id,
client_id,
)
select_vllm_module = current_config["selected_module"].get("VLLM")
if not select_vllm_module:
raise ValueError("您还未设置默认的视觉分析模块")
vllm_type = (
select_vllm_module
if "type" not in current_config["VLLM"][select_vllm_module]
else current_config["VLLM"][select_vllm_module]["type"]
)
if not vllm_type:
raise ValueError(f"无法找到VLLM模块对应的供应器{vllm_type}")
vllm = create_instance(
vllm_type, current_config["VLLM"][select_vllm_module]
)
response = vllm.response(question, image_base64)
return_json = {
"success": True,
@@ -99,7 +153,13 @@ class VisionHandler:
"""处理 MCP Vision GET 请求"""
try:
vision_explain = get_vision_url(self.config)
message = f"MCP Vision 接口运行正常，视觉解释接口地址是：{vision_explain}"
if vision_explain and len(vision_explain) > 0 and "null" != vision_explain:
message = (
f"MCP Vision 接口运行正常，视觉解释接口地址是：{vision_explain}"
)
else:
message = "MCP Vision 接口运行不正常，请打开data目录下的.config.yaml文件，找到【server.vision_explain】，设置好地址"
response = web.Response(text=message, content_type="text/plain")
except Exception as e:
self.logger.bind(tag=TAG).error(f"MCP Vision GET请求异常: {e}")
main/xiaozhi-server/core/connection.py
+4
View File
@@ -478,6 +478,8 @@ class ConnectionHandler:
self.memory = modules["memory"]
def _initialize_memory(self):
if self.memory is None:
return
"""初始化记忆模块"""
self.memory.init_memory(
role_id=self.device_id,
@@ -518,6 +520,8 @@ class ConnectionHandler:
self.logger.bind(tag=TAG).info("使用主LLM作为意图识别模型")
def _initialize_intent(self):
if self.intent is None:
return
self.intent_type = self.config["Intent"][
self.config["selected_module"]["Intent"]
]["type"]
main/xiaozhi-server/core/handle/mcpHandle.py
+13 -9
View File
@@ -2,6 +2,7 @@ import json
import asyncio
from concurrent.futures import Future
from core.utils.util import get_vision_url
from core.utils.auth import AuthToken
TAG = __name__
@@ -207,15 +208,18 @@ async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict):
async def send_mcp_initialize_message(conn):
"""发送MCP初始化消息"""
# 智控台暂时不启动视觉分析
if conn.read_config_from_api:
vision = {}
else:
vision_url = get_vision_url(conn.config)
vision = {
"url": vision_url,
"token": "test_token",
}
vision_url = get_vision_url(conn.config)
# 密钥生成token
auth = AuthToken(conn.config["server"]["auth_key"])
token = auth.generate_token(conn.headers.get("device-id"))
vision = {
"url": vision_url,
"token": token,
}
conn.logger.bind(tag=TAG).info(f"视觉服务信息: {vision}")
payload = {
"jsonrpc": "2.0",
main/xiaozhi-server/core/http_server.py
+13 -4
View File
@@ -35,16 +35,25 @@ class SimpleHttpServer:
async def start(self):
server_config = self.config["server"]
host = server_config.get("ip", "0.0.0.0")
port = int(server_config.get("http_port"))
port = int(server_config.get("http_port", 8003))
if port:
app = web.Application()
read_config_from_api = server_config.get("read_config_from_api", False)
if not read_config_from_api:
# 如果没有开启智控台，只是单模块运行，就需要再添加简单OTA接口，用于下发websocket接口
app.add_routes(
[
web.get("/xiaozhi/ota/", self.ota_handler.handle_get),
web.post("/xiaozhi/ota/", self.ota_handler.handle_post),
web.options("/xiaozhi/ota/", self.ota_handler.handle_post),
]
)
# 添加路由
app.add_routes(
[
web.get("/xiaozhi/ota/", self.ota_handler.handle_get),
web.post("/xiaozhi/ota/", self.ota_handler.handle_post),
web.options("/xiaozhi/ota/", self.ota_handler.handle_post),
web.get("/mcp/vision/explain", self.vision_handler.handle_get),
web.post("/mcp/vision/explain", self.vision_handler.handle_post),
web.options("/mcp/vision/explain", self.vision_handler.handle_post),
main/xiaozhi-server/core/utils/auth.py
+43
View File
@@ -0,0 +1,43 @@
import jwt
import time
from datetime import datetime, timedelta, timezone
from typing import Optional, Tuple
class AuthToken:
def __init__(self, secret_key: str):
self.secret_key = secret_key
def generate_token(self, device_id: str) -> str:
"""
生成JWT token
:param device_id: 设备ID
:return: JWT token字符串
"""
# 设置过期时间为1小时后
expire_time = datetime.now(timezone.utc) + timedelta(hours=1)
# 创建payload
payload = {"device_id": device_id, "exp": expire_time.timestamp()}
# 使用JWT进行编码
token = jwt.encode(payload, self.secret_key, algorithm="HS256")
return token
def verify_token(self, token: str) -> Tuple[bool, Optional[str]]:
"""
验证token
:param token: JWT token字符串
:return: (是否有效, 设备ID)
"""
try:
# 解码token
payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
# 检查是否过期
if payload["exp"] < time.time():
return False, None
return True, payload["device_id"]
except jwt.InvalidTokenError:
return False, None
main/xiaozhi-server/requirements.txt
+1
View File
@@ -31,3 +31,4 @@ chardet==5.2.0
aioconsole==0.8.1
markitdown==0.1.1
mcp-proxy==0.6.0
PyJWT==2.8.0