hermes_api_stub.py
6.13 KB
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
#!/usr/bin/env python3
"""
Hermes API Stub — 轻量级 HTTP SSE 服务
模拟 lemonPCRM /api/master/chat 端点，内部调用 Kimi API。
Usage:
python3 hermes_api_stub.py # 前台运行
python3 hermes_api_stub.py --daemon # 后台运行
"""
import json, os, sys, argparse, urllib.request, urllib.error
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
PORT = 3010
KIMI_API_KEY = ""
KIMI_BASE_URL = "https://api.kimi.com/coding"
KIMI_MODEL = "kimi-for-coding"
def load_kimi_key():
"""Try to load Kimi API key from hermes-agent config or env."""
global KIMI_API_KEY
# Try hermes-agent config
try:
cfg_path = os.path.expanduser("~/.hermes/config.yaml")
if os.path.exists(cfg_path):
import yaml
with open(cfg_path) as f:
cfg = yaml.safe_load(f)
providers = cfg.get("providers", {})
kimi = providers.get("kimi", {})
KIMI_API_KEY = kimi.get("api_key", "")
if KIMI_API_KEY:
print(f"[STUB] Loaded Kimi key from {cfg_path}")
return
except Exception:
pass
# Try env
KIMI_API_KEY = os.environ.get("KIMI_API_KEY", "")
if KIMI_API_KEY:
print("[STUB] Loaded Kimi key from env")
def call_kimi(messages, timeout=60):
"""Call Kimi chat completions API. Returns generator of chunks."""
payload = json.dumps({
"model": KIMI_MODEL,
"messages": messages,
"stream": True
}).encode("utf-8")
req = urllib.request.Request(
f"{KIMI_BASE_URL}/v1/chat/completions",
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {KIMI_API_KEY}"
},
method="POST"
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
for line in resp:
line = line.decode("utf-8", errors="replace").strip()
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
try:
obj = json.loads(data)
delta = obj.get("choices", [{}])[0].get("delta", {})
content = delta.get("content", "")
if content:
yield content
except json.JSONDecodeError:
pass
except Exception as e:
yield f"\n[ERROR: {e}]"
class Handler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
print(f"[STUB] {self.address_string()} - {format % args}")
def do_GET(self):
if self.path == "/api/health":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": "ok", "stub": True}).encode())
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
if self.path == "/api/master/chat":
self._handle_chat()
else:
self.send_response(404)
self.end_headers()
def _handle_chat(self):
# Read body
body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
try:
data = json.loads(body.decode("utf-8"))
except Exception:
self.send_response(400)
self.end_headers()
return
message = data.get("message", "")
session_id = data.get("sessionId", f"session-{id(data)}")
print(f"[STUB] Chat request session={session_id} msg_len={len(message)}")
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "keep-alive")
self.end_headers()
def send_event(event, data_obj):
chunk = f"event: {event}\ndata: {json.dumps(data_obj)}\n\n"
self.wfile.write(chunk.encode("utf-8"))
self.wfile.flush()
# session event
send_event("session", {"sessionId": session_id})
send_event("progress", {"step": "initializing", "message": "🤖 Hermes Stub 启动中..."})
if not KIMI_API_KEY:
send_event("error", {"message": "Kimi API key not configured"})
return
messages = [
{"role": "system", "content": "你是一个微信智能助手。根据收到的消息，给出简短、自然、得体的中文回复。不要加任何解释或格式标记。"},
{"role": "user", "content": message}
]
full_reply = []
for chunk in call_kimi(messages):
if chunk.startswith("\n[ERROR:"):
send_event("error", {"message": chunk[8:-1]})
return
full_reply.append(chunk)
send_event("agent_output", {"agent": "master", "message": chunk, "streamType": "agent_output"})
reply_text = "".join(full_reply)
send_event("result", {
"sessionId": session_id,
"type": "text",
"output": reply_text,
"toolCalls": [],
"rounds": 1
})
class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
allow_reuse_address = True
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=PORT)
parser.add_argument("--daemon", action="store_true")
args = parser.parse_args()
load_kimi_key()
if args.daemon:
pid = os.fork()
if pid > 0:
print(f"[STUB] Daemon started (PID={pid})")
sys.exit(0)
server = ThreadedHTTPServer(("0.0.0.0", args.port), Handler)
print(f"[STUB] Hermes API Stub listening on http://0.0.0.0:{args.port}")
print(f"[STUB] Kimi key loaded: {'yes' if KIMI_API_KEY else 'NO — set KIMI_API_KEY env or check ~/.hermes/config.yaml'}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n[STUB] Shutting down")
server.shutdown()
if __name__ == "__main__":
main()