Skills › Scientific Agent Skills › autoskill › backends.py
backends.py
140 lines · skills/autoskill/scripts/backends.py at 5f2bc03
1import ipaddress
2import os
3import sys
4from urllib.parse import urlparse
6import httpx
9def _is_loopback(host):
10 if host == "localhost":
11 return True
12 try:
13 return ipaddress.ip_address(host).is_loopback
14 except ValueError:
15 return False
18def check_remote_endpoint(endpoint, label):
19 """Reject cleartext transport to a remote host, and name the destination.
21 This backend sends summaries derived from the user's screen-capture history.
22 The endpoint is read from config.yaml, so it is worth being explicit about
23 where that data is about to go, and refusing to send it -- along with an API
24 key header -- over plaintext HTTP to anything but the local machine.
25 """
26 parsed = urlparse(endpoint)
27 host = parsed.hostname or ""
29 if (parsed.scheme not in ("http", "https") or not host
30 or parsed.username is not None or parsed.password is not None
31 or parsed.query or parsed.fragment):
32 raise ValueError(
33 f"{label} endpoint must be an http:// or https:// base URL with a host, "
34 "without credentials, query, or fragment"
35 )
37 if parsed.scheme == "http" and not _is_loopback(host):
38 raise ValueError(
39 f"{label} endpoint {endpoint!r} uses plaintext HTTP to a remote host. "
40 "Screen-derived content and your API key would cross the network "
41 "unencrypted. Use https://, or point the endpoint at localhost."
42 )
44 if not _is_loopback(host):
45 print(
46 f"[autoskill] sending screen-derived summaries to {parsed.scheme}://{host}",
47 file=sys.stderr,
48 )
50 return endpoint
53class ClaudeBackend:
54 def __init__(self, api_key, model, client=None):
55 self.api_key = api_key
56 self.model = model
57 self.client = client or httpx.Client(base_url="https://api.anthropic.com", timeout=60.0)
59 def __call__(self, prompt):
60 response = self.client.post(
61 "/v1/messages",
62 headers={
63 "x-api-key": self.api_key,
64 "anthropic-version": "2023-06-01",
65 "content-type": "application/json",
66 },
67 json={
68 "model": self.model,
69 "max_tokens": 4096,
70 "messages": [{"role": "user", "content": prompt}],
71 },
72 )
73 response.raise_for_status()
74 payload = response.json()
75 reason = payload.get("stop_reason")
76 if reason not in ("end_turn", "stop_sequence"):
77 raise RuntimeError(f"Claude synthesis did not finish normally: {reason}")
78 text = "".join(block["text"] for block in payload.get("content", [])
79 if block.get("type") == "text" and isinstance(block.get("text"), str))
80 if not text.strip():
81 raise RuntimeError("Claude synthesis returned no text content")
82 return text
85class LocalBackend:
86 def __init__(self, endpoint, model, client=None, api_key=None):
87 self.endpoint = endpoint
88 self.model = model
89 self.api_key = api_key
90 self.client = client or httpx.Client(base_url=endpoint, timeout=120.0)
92 def __call__(self, prompt):
93 response = self.client.post(
94 "/chat/completions",
95 headers={"Authorization": f"Bearer {self.api_key}"} if self.api_key else {},
96 json={
97 "model": self.model,
98 "messages": [{"role": "user", "content": prompt}],
99 "stream": False,
100 },
101 )
102 response.raise_for_status()
103 payload = response.json()
104 choices = payload.get("choices", [])
105 if not choices:
106 raise RuntimeError("Local synthesis returned no choices")
107 choice = choices[0]
108 reason = choice.get("finish_reason")
109 if reason != "stop":
110 raise RuntimeError(f"Local synthesis did not finish normally: {reason}")
111 text = choice.get("message", {}).get("content")
112 if not isinstance(text, str) or not text.strip():
113 raise RuntimeError("Local synthesis returned no text content")
114 return text
117def make_backend(config):
118 kind = config.get("backend")
119 if kind == "claude":
120 api_key = os.environ.get("ANTHROPIC_API_KEY")
121 if not api_key:
122 raise RuntimeError("ANTHROPIC_API_KEY environment variable not set")
123 model = config.get("claude", {}).get("model", "claude-opus-4-7")
124 return ClaudeBackend(api_key=api_key, model=model)
126 if kind == "foundry":
127 api_key = os.environ.get("FOUNDRY_API_KEY")
128 if not api_key:
129 raise RuntimeError("FOUNDRY_API_KEY environment variable not set")
130 f = config.get("foundry", {})
131 endpoint = check_remote_endpoint(f["endpoint"], "foundry")
132 client = httpx.Client(base_url=endpoint, timeout=60.0)
133 return ClaudeBackend(api_key=api_key, model=f.get("model", "claude-opus-4-7"), client=client)
135 if kind == "local":
136 l = config.get("local", {})
137 return LocalBackend(endpoint=check_remote_endpoint(l["endpoint"], "local"), model=l["model"],
138 api_key=os.environ.get("LM_API_TOKEN"))
140 raise ValueError(f"unknown backend: {kind!r}")