mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 10:59:59 +00:00
kevinveenbirkenbachandClaude Opus 5
dfe7a7fdd4
The merged branch moved every role-local script out of files/ into a folder named for its language (files/python/, files/php/, files/shell/), but the moves lost their staging and never landed. HEAD still carried the old paths while the worktree had the new ones. Move the 24 remaining scripts and update the six places that resolve them: the mcp-adapter instance/isolation/probe tasks, the litellm key task, the adapter README and the endpoint-path lint. Every changed line in those files is a path. Also drop roles/sys-svc-compose/files/swarm/stack_ready.sh, a stale duplicate of the copy already tracked under files/shell/swarm/. The added comments are the branch's own `nocheck: mirrored-unit-test` directives, which state why a script whose measurement only exists inside its container cannot have a mirrored unit test. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
# nocheck: mirrored-unit-test - reads MCP_BASE_URL and the endpoint key from the
|
|
# container environment at import and then drives a live SSE handshake; the module
|
|
# cannot be loaded outside the running Baserow container
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
base_url = os.environ["MCP_BASE_URL"].rstrip("/")
|
|
endpoint_path = os.environ["MCP_ENDPOINT_PATH"]
|
|
key = os.environ["MCP_ENDPOINT_KEY"]
|
|
|
|
TIMEOUT = 15
|
|
STREAM_LINES = 4
|
|
|
|
TRANSIENT_STATUS = frozenset({0, 429, 502, 503, 504})
|
|
|
|
|
|
def refuse(message, status):
|
|
marker = "PENDING" if status in TRANSIENT_STATUS else "REJECTED"
|
|
sys.stderr.write(f"{marker} {message}\n")
|
|
sys.exit(1)
|
|
|
|
|
|
def probe(candidate):
|
|
"""Return (status, content_type, first stream lines) of an SSE GET.
|
|
|
|
Args:
|
|
candidate: endpoint key placed in the URL path.
|
|
"""
|
|
url = f"{base_url}{endpoint_path}/{candidate}/sse"
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=TIMEOUT) as response: # noqa: S310 fixed internal http origin
|
|
lines = [response.readline() for _ in range(STREAM_LINES)]
|
|
return response.status, response.headers.get("Content-Type", ""), lines
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, error.headers.get("Content-Type", ""), []
|
|
except OSError:
|
|
return 0, "", []
|
|
|
|
|
|
status, content_type, _ = probe("0" * len(key))
|
|
if status != 401:
|
|
refuse(f"unauthenticated probe answered {status} {content_type}", status)
|
|
|
|
status, content_type, lines = probe(key)
|
|
if status != 200 or "text/event-stream" not in content_type:
|
|
refuse(f"authenticated probe answered {status} {content_type}", status)
|
|
|
|
stream = b"".join(lines)
|
|
if b"event: endpoint" not in stream:
|
|
refuse(f"authenticated probe streamed {stream!r}", None)
|
|
|
|
print("OK")
|