OpenCode、OpenClaw、Hermes对外APIs接口

def jiexi(response):    print(response.text)    data = response.json() if isinstance(response.json(), list) else [response.json()]    for r in data:        role = r['info']['role']        if role == 'user':            print(f"{role}：{r['parts'][0]['text']}")        elif role == 'assistant':            for p in r['parts']:                if p["type"] == "reasoning":#思考                    pass                elif p["type"] == "text":                    print(f"{role}：{p['text']}")res1 = requests.get(    url='http://127.0.0.1:4096/session/ses_idxxxxxx/message',)#{"name":"NotFoundError","data":{"message":"Session not found: ses_116935062ffejURD7xVWvGDax"}}if "Session not found" in res1.text:    pass    # 需要重新创建会话else:    jiexi(res1)

import asynciofrom contextlib import asynccontextmanagerfrom typing import Optionalimport httpxfrom fastapi import FastAPI, HTTPExceptionfrom loguru import loggerfrom pydantic import BaseModel# ==================== 日志配置 ====================# 移除默认 handler，避免重复打印logger.remove()# 1. 控制台输出：带颜色，仅显示 INFO 及以上logger.add(    lambda msg: print(msg, end=""),  # 兼容 Windows 控制台颜色    level="INFO",    format="<green>{time:HH:mm:ss}</green> | <level>{level:<7}</level> | {message}",    colorize=True)# 2. 文件输出：按天切割，保留30天，记录完整内容logger.add(    "logs/opencode_{time:YYYY-MM-DD}.log",    rotation="00:00",       # 每天零点切割    retention="30 days",    # 保留30天    encoding="utf-8",    level="DEBUG",    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level:<7} | {message}",    enqueue=True            # 🔑 异步写入，不阻塞 FastAPI 事件循环)# ==================== 数据模型 ====================class ChatRequest(BaseModel):    phone: str    query: strclass ChatResponse(BaseModel):    phone: str    session_id: str    reply: str# ==================== OpenCode 客户端 ====================class OpenCodeClient:    # 🔧 在此处定义系统提示词，方便后续维护    SYSTEM_PROMPT = (        "你是一个专业的智能助手。请始终使用中文回答，"        "保持语气友好、专业，并在回答前仔细分析用户的问题。"    )    def __init__(self, base_url: str = "http://127.0.0.1:4096"):        self.base_url = base_url.rstrip("/")        self.phone_session_map: dict[str, str] = {}        self._lock = asyncio.Lock()        self._client: Optional[httpx.AsyncClient] = None    async def start(self):        self._client = httpx.AsyncClient(timeout=60*10)        logger.info("OpenCode HTTP 客户端已初始化")    async def stop(self):        if self._client:            await self._client.aclose()        logger.info("OpenCode HTTP 客户端已关闭")    async def _create_session(self, phone: str) -> str:        # 1. 创建新会话        res = await self._client.post(f"{self.base_url}/session")        res.raise_for_status()        session_id = res.json()["id"]        logger.info(f"[新建会话] phone={phone} | session_id={session_id}")        # 🔑 2. 首次创建时，立即注入系统提示词        sys_payload = {            "parts": [{"type": "text", "text": self.SYSTEM_PROMPT}]        }        sys_res = await self._client.post(            f"{self.base_url}/session/{session_id}/message",            json=sys_payload,            headers={'Content-Type': 'application/json'}        )        if sys_res.status_code == 200:            logger.debug(f"[系统提示词注入成功] phone={phone} | session={session_id}")        else:            logger.warning(                f"[系统提示词注入失败] phone={phone} | session={session_id} | "                f"status={sys_res.status_code} | body={sys_res.text}"            )        return session_id    async def _session_exists(self, session_id: str) -> bool:        res = await self._client.get(f"{self.base_url}/session/{session_id}/message")        return "Session not found" not in res.text    async def get_or_create_session(self, phone: str) -> str:        async with self._lock:            session_id = self.phone_session_map.get(phone)            if session_id:                if await self._session_exists(session_id):                    logger.debug(f"[复用会话] phone={phone} | session_id={session_id}")                    return session_id                else:                    logger.warning(f"[会话失效] phone={phone} | old_session={session_id}")                    del self.phone_session_map[phone]            new_session_id = await self._create_session(phone)            self.phone_session_map[phone] = new_session_id            return new_session_id    @staticmethod    def _extract_reply(data) -> str:        items = data if isinstance(data, list) else [data]        texts = []        for r in items:            if r.get("info", {}).get("role") == "assistant":                for p in r.get("parts", []):                    if p.get("type") == "text":                        texts.append(p["text"])        return "\n".join(texts)    async def chat(self, phone: str, query: str) -> dict:        # 📝 记录用户发送内容        logger.info(f"[用户提问] phone={phone} | query={query!r}")        session_id = await self.get_or_create_session(phone)        payload = {"parts": [{"type": "text", "text": query}]}        res = await self._client.post(            f"{self.base_url}/session/{session_id}/message",            json=payload,        )        # 极端情况重试        if "Session not found" in res.text:            logger.warning(f"[会话异常重试] phone={phone} | session={session_id}")            async with self._lock:                self.phone_session_map.pop(phone, None)            session_id = await self.get_or_create_session(phone)            res = await self._client.post(                f"{self.base_url}/session/{session_id}/message",                json=payload,            )        reply = self._extract_reply(res.json())        # 📝 记录 AI 回复内容        logger.info(f"[AI回复] phone={phone} | session={session_id} | reply={reply!r}")        return {"session_id": session_id, "reply": reply}# ==================== FastAPI 应用 ====================opencode = OpenCodeClient()@asynccontextmanagerasync def lifespan(app: FastAPI):    await opencode.start()    yield    await opencode.stop()app = FastAPI(title="OpenCode Proxy", lifespan=lifespan)@app.post("/chat", response_model=ChatResponse)async def chat(req: ChatRequest):    try:        result = await opencode.chat(phone=req.phone, query=req.query)    except httpx.HTTPStatusError as e:        logger.error(f"[OpenCode错误] phone={req.phone} | status={e.response.status_code} | detail={e.response.text}")        raise HTTPException(status_code=e.response.status_code, detail=f"OpenCode 请求失败: {e.response.text}")    except Exception as e:        logger.exception(f"[未知异常] phone={req.phone} | error={e}")        raise HTTPException(status_code=500, detail=str(e))    return ChatResponse(        phone=req.phone,        session_id=result["session_id"],        reply=result["reply"],    )

curl http://localhost:8080/v1/responses \  -H "Authorization: Bearer YOUR_TOKEN" \  -H "Content-Type: application/json" \  -d '{    "model": "openclaw",    "input": "你好"  }'

curl http://localhost:8080/v1/responses \  -H "Authorization: Bearer YOUR_TOKEN" \  -H "Content-Type: application/json" \  -d '{    "model": "openclaw",    "user": "alice",    "input": "你好"  }'

curl http://localhost:8080/v1/responses \  -H "Authorization: Bearer YOUR_TOKEN" \  -H "x-openclaw-session-key: customer-9527" \  -H "Content-Type: application/json" \  -d '{    "model": "openclaw",    "input": "你好"  }'

API_SERVER_ENABLED=trueAPI_SERVER_KEY=change-me-local-dev# Optional: only if a browser must call Hermes directly# API_SERVER_CORS_ORIGINS=http://localhost:3000

hermes gateway

[API Server] API server listening on http://127.0.0.1:8642

# Test with curlcurl http://localhost:8642/v1/chat/completions \  -H "Authorization: Bearer change-me-local-dev" \  -H "Content-Type: application/json" \  -d '{"model": "hermes-agent", "messages": [{"role": "user", "content": "Hello!"}]}'