mirror of
https://github.com/patriceckhart/zot.git
synced 2026-06-27 13:56:33 +02:00
a670442c9e
two new ways to embed the zot agent runtime in third-party apps:
1. pkg/zotcore - public Go SDK
- Runtime type: New(Config), Prompt(ctx,text,imgs)->chan Event,
Cancel, Compact, SetModel, State, Messages, Cost, ListModels,
Close. Concurrent-safe; one prompt at a time per Runtime,
ErrBusy if you try to overlap. Spawn multiple Runtimes for
multiple projects.
- Public types mirror the JSON-RPC wire schema 1:1 so consumers
can share parsing code with the out-of-process clients.
- Internal core/agent/provider stay internal; SDK is a thin
facade that exposes only what's stable.
2. zot rpc subcommand - newline-delimited JSON on stdin/stdout
- 'zot rpc' (or 'zot --rpc') turns the agent runtime into a
subprocess that any language can drive via pipes.
- Commands: hello, prompt, abort, compact, get_state,
get_messages, clear, set_model, get_models, ping. Each
optionally carries an id; the matching response echoes it.
- Stream notifications: turn_start, user_message,
assistant_start, text_delta, tool_call, tool_progress,
tool_result, assistant_message, usage, turn_end, done,
error, compact_done. Same shape as the existing --json mode
events (modes.EventToJSON / ContentToJSON were exported
for reuse).
- Auth: optional ZOTCORE_RPC_TOKEN env var; first command
must be hello {token: ...} when set. Without the env var
the spawning process is implicitly trusted.
- Concurrency: one prompt or compact at a time per process,
enforced by a turnMu mutex. abort fires immediately
regardless. Stdin close exits the process.
3. docs/rpc.md - full schema reference
4. examples/rpc/{python,node,shell,go} - reference clients
5. examples/sdk - in-process Go embedding example
6. README updated with a new modes entry and an embedding section
114 lines
3.3 KiB
Python
114 lines
3.3 KiB
Python
"""Minimal Python client for the `zot rpc` JSON protocol.
|
|
|
|
Usage:
|
|
python zot_client.py "fix the failing test"
|
|
|
|
Spawns `zot rpc`, sends one prompt, prints assistant text as it streams,
|
|
and exits when the turn finishes. No external dependencies — stdlib
|
|
only. Implements just enough of the protocol to be useful as a
|
|
starting point; see docs/rpc.md for the full schema.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import uuid
|
|
|
|
|
|
class ZotClient:
|
|
def __init__(self, *flags: str) -> None:
|
|
argv = ["zot", "rpc", *flags]
|
|
env = os.environ.copy()
|
|
self.proc = subprocess.Popen(
|
|
argv,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=sys.stderr,
|
|
env=env,
|
|
text=True,
|
|
bufsize=1, # line-buffered
|
|
)
|
|
self._lock = threading.Lock()
|
|
|
|
def send(self, **command) -> str:
|
|
"""Send one command, return its `id`."""
|
|
if "id" not in command:
|
|
command["id"] = uuid.uuid4().hex[:8]
|
|
line = json.dumps(command)
|
|
with self._lock:
|
|
assert self.proc.stdin is not None
|
|
self.proc.stdin.write(line + "\n")
|
|
self.proc.stdin.flush()
|
|
return command["id"]
|
|
|
|
def events(self):
|
|
"""Yield every JSON object the server emits, until stdout closes."""
|
|
assert self.proc.stdout is not None
|
|
for line in self.proc.stdout:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
yield json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
def close(self) -> None:
|
|
if self.proc.stdin is not None:
|
|
try:
|
|
self.proc.stdin.close()
|
|
except Exception:
|
|
pass
|
|
self.proc.wait(timeout=5)
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 2:
|
|
print("usage: zot_client.py <prompt>", file=sys.stderr)
|
|
return 2
|
|
prompt = " ".join(sys.argv[1:])
|
|
|
|
client = ZotClient()
|
|
try:
|
|
token = os.environ.get("ZOTCORE_RPC_TOKEN")
|
|
if token:
|
|
client.send(type="hello", token=token)
|
|
|
|
client.send(type="prompt", message=prompt)
|
|
|
|
for ev in client.events():
|
|
t = ev.get("type")
|
|
if t == "text_delta":
|
|
sys.stdout.write(ev.get("delta", ""))
|
|
sys.stdout.flush()
|
|
elif t == "tool_call":
|
|
print(
|
|
f"\n[tool] {ev.get('name')}({json.dumps(ev.get('args', {}))})",
|
|
file=sys.stderr,
|
|
)
|
|
elif t == "tool_result":
|
|
if ev.get("is_error"):
|
|
print("[tool error]", file=sys.stderr)
|
|
elif t == "usage":
|
|
cum = ev.get("cumulative", {})
|
|
print(
|
|
f"\n[usage] cum input={cum.get('input')} "
|
|
f"output={cum.get('output')} cost=${cum.get('cost_usd', 0):.4f}",
|
|
file=sys.stderr,
|
|
)
|
|
elif t == "done":
|
|
break
|
|
elif t == "error":
|
|
print(f"\n[error] {ev.get('message')}", file=sys.stderr)
|
|
return 1
|
|
print()
|
|
return 0
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|