a digital entity named phi that roams bsky
phi.zzstoatzz.io
surface valid relay names to the LLM via a dynamic system prompt
uses the same DI pattern phi already has for identity / notifications /
memory: a @agent.system_prompt(dynamic=True) that injects a [KNOWN RELAYS]
block with the current hostnames, cached 5 min from the snapshot endpoint.
the LLM now sees the valid values upfront and passes exact hostnames to
check_relays(name=...) without guessing.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
+42
-6
+10
src/bot/agent.py
+10
src/bot/agent.py
···
174
174
return f"[NOW]: {now.strftime('%Y-%m-%d %H:%M UTC')}"
175
175
176
176
@self.agent.system_prompt(dynamic=True)
177
+
async def inject_known_relays() -> str:
178
+
"""List the valid relay hostnames for check_relays(name=...)."""
179
+
from bot.tools.bluesky import fetch_relay_names
180
+
181
+
names = await fetch_relay_names()
182
+
if not names:
183
+
return ""
184
+
return "[KNOWN RELAYS]: " + ", ".join(names)
185
+
186
+
@self.agent.system_prompt(dynamic=True)
177
187
def inject_notifications(ctx: RunContext[PhiDeps]) -> str:
178
188
"""Render the notifications batch as the [NEW NOTIFICATIONS] block."""
179
189
return _format_notifications_block(ctx.deps.notifications_context or {})
+32
-6
src/bot/tools/bluesky.py
+32
-6
src/bot/tools/bluesky.py
···
3
3
import asyncio
4
4
import ipaddress
5
5
import socket
6
+
import time
6
7
from datetime import date
7
8
from typing import Annotated
8
9
from urllib.parse import urlparse
···
15
16
from bot.core.atproto_client import bot_client
16
17
from bot.core.mentionable import add_handle, get_mentionable_handles, remove_handle
17
18
from bot.tools._helpers import PhiDeps, _check_services_impl, _is_owner, _relative_age
19
+
20
+
# cached relay names, refreshed from the snapshot endpoint. surfaced to
21
+
# the LLM via a dynamic system prompt so it picks from real values when
22
+
# calling check_relays(name=...).
23
+
_RELAY_NAMES_TTL = 300 # 5 minutes
24
+
_relay_names_cache: dict = {"names": [], "fetched_at": 0.0}
25
+
26
+
27
+
async def fetch_relay_names() -> list[str]:
28
+
now = time.time()
29
+
if (
30
+
now - _relay_names_cache["fetched_at"] < _RELAY_NAMES_TTL
31
+
and _relay_names_cache["names"]
32
+
):
33
+
return _relay_names_cache["names"]
34
+
try:
35
+
async with httpx.AsyncClient(timeout=10) as http:
36
+
r = await http.get(settings.monitors_url)
37
+
r.raise_for_status()
38
+
names = sorted({m.get("name", "") for m in r.json() if m.get("name")})
39
+
_relay_names_cache["names"] = names
40
+
_relay_names_cache["fetched_at"] = now
41
+
return names
42
+
except Exception:
43
+
return _relay_names_cache["names"] # fall back to last known
18
44
19
45
20
46
def register(agent):
···
180
206
) -> str:
181
207
"""Check the atproto relay fleet nate evaluates via relay-eval.
182
208
183
-
Default (no name): current snapshot of every relay, grouped by
184
-
status — answers "how's the fleet right now." Report headlines
185
-
verbatim.
209
+
Default (no name): current snapshot of every relay, grouped by status.
210
+
Report headlines verbatim.
186
211
187
-
With name: recent coverage history for one relay (summary stats +
188
-
recent points) — answers "what was X's coverage yesterday."
212
+
With name: recent coverage history for one relay — summary stats +
213
+
recent points. Valid hostnames are listed in the [KNOWN RELAYS]
214
+
system-prompt block; pass one of those exactly.
189
215
190
216
For app health (plyr, PDS, prefect, etc), use check_services."""
191
217
if name:
···
204
230
points = data.get("points", [])
205
231
summary = data.get("summary", {})
206
232
if not points:
207
-
return f"no history found for {name}"
233
+
return f"no history found for '{name}'"
208
234
209
235
mean = summary.get("mean_coverage_pct", 0)
210
236
lo = summary.get("min_coverage_pct", 0)