Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174"""Real daemon MCP extension: stdio discovery, namespaced calls, env scrubbing, teardown.
The fake server speaks line-delimited JSON-RPC over a real subprocess, so thisexercises the actual client library and launcher instead of a stub inside the daemon."""import contextlibimport http.serverimport jsonimport osfrom pathlib import Pathimport subprocessimport sysimport tempfileimport threadingimport timeimport urllib.errorimport urllib.request
ROOT = Path(__file__).resolve().parents[2]SERVER = Path(__file__).resolve().parent/"fake_mcp_server.py"
class Provider(http.server.BaseHTTPRequestHandler): requests = []
def log_message(self, *_): pass
def do_POST(self): request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) self.requests.append(request) tools = sorted(tool["name"] for tool in request.get("tools", [])) answered = any(item.get("type") == "function_call_output" for item in request["input"]) mcp_tool = next((name for name in tools if name.startswith("mcp_")), None) if mcp_tool and not answered: output = [{"type": "function_call", "id": "fc1", "call_id": "call-mcp", "name": mcp_tool, "arguments": json.dumps({"message": "ping from albedo"}), "status": "completed"}] else: output = [{"type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "done", "annotations": []}]}] body = ("data: " + json.dumps({"type": "response.completed", "response": { "id": "fixture", "status": "completed", "output": output, "usage": {"input_tokens": 12, "output_tokens": 1}, }}) + "\n\n").encode() self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
def run(endpoint): with tempfile.TemporaryDirectory(prefix="albedo-mcp-") as directory: root = Path(directory) home, workspace, user_home = root/"state", root/"workspace", root/"user" for path in (home, workspace, user_home): path.mkdir(mode=0o700) closed = root/"closed" (home/"config.json").write_text(json.dumps({"active": "fixture", "providers": {"fixture": { "baseUrl": endpoint, "apiKey": "fixture-key", "model": "fixture", "protocol": "responses", }}}))
def configure(args, startup=20000): (home/"extensions.json").write_text(json.dumps({"models": {"refreshHours": 0}, "mcp": {"servers": {"fake": { "type": "stdio", "command": sys.executable, "args": args, "env": {"FAKE_SECRET": {"env": "ALBEDO_MCP_SECRET"}, "FAKE_MCP_CLOSED": {"env": "ALBEDO_MCP_CLOSED"}}, "startupTimeoutMs": startup, }}}}))
configure([str(SERVER)]) credentials = home/"mcp-credentials.json" credentials.write_text(json.dumps({"servers":{"fake":{"env":{"FAKE_SECRET":"stored-secret"}}}})) credentials.chmod(0o600) env = dict(os.environ, HOME=str(user_home), ALBEDO_HOME=str(home), ALBEDO_PARENT_PID=str(os.getpid()), ALBEDO_MCP_SECRET="configured-secret", ALBEDO_MCP_CLOSED=str(closed), ALBEDO_MCP_AMBIENT="must-not-reach-the-server") connection = None
def cli(*args): result = subprocess.run(["node", "cli/bin/albedo.mjs", *args], cwd=ROOT, env=env, capture_output=True, text=True, timeout=60) assert result.returncode == 0, result.stdout + result.stderr + (home/"daemon.log").read_text() return result.stdout
def api(path, data=None): request = urllib.request.Request(f"http://127.0.0.1:{connection['port']}" + path, headers={"Authorization": "Bearer " + connection["token"], "Content-Type": "application/json"}, data=None if data is None else json.dumps(data).encode()) with urllib.request.urlopen(request, timeout=60) as response: return json.load(response)
def ready(session): deadline = time.monotonic() + 60 while time.monotonic() < deadline: if not api(f"/sessions/{session}/status")["running"]: return time.sleep(.025) raise AssertionError("session did not settle: " + (home/"daemon.log").read_text())
def turn(session, prompt): before = len(Provider.requests) api(f"/sessions/{session}/events", {"content": prompt}) ready(session) return Provider.requests[before:]
try: cli("sessions") connection = json.loads((home/"daemon.json").read_text()) session = json.loads(cli("new", str(workspace)))["session"] route = f"/sessions/{session}/extensions" enabled = api(route, {"name": "mcp", "enabled": True}) mcp = next(item for item in enabled if item["name"] == "mcp") assert mcp["enabled"] and mcp["plugins"] == ["managed"], mcp
requests = turn(session, "use the mcp server") assert len(requests) == 2, requests context = "\n".join(item["content"] for item in requests[0]["input"] if isinstance(item.get("content"), str) and item["content"].startswith("<extension-context")) assert "name=\"mcp\"" in context and "echo" in context, context advertised = [tool["name"] for tool in requests[0]["tools"] if tool["name"].startswith("mcp_")] assert len(advertised) == 1 and advertised[0].startswith("mcp_fake_echo_"), advertised result = next(item["output"] for item in requests[1]["input"] if item.get("type") == "function_call_output") assert json.loads(json.loads(result)["content"][0]["text"]) == { "echoed": "ping from albedo", "secret": "stored-secret", "ambient": None}, result summary = next(item for item in api(route) if item["name"] == "mcp") assert summary["tools"] == advertised, summary
disabled = api(route, {"name": "mcp", "enabled": False}) assert not next(item["enabled"] for item in disabled if item["name"] == "mcp") deadline = time.monotonic() + 15 while time.monotonic() < deadline and not closed.exists(): time.sleep(.05) assert closed.exists(), "disabling mcp must close its server process" requests = turn(session, "the server is gone") assert not any(tool["name"].startswith("mcp_") for tool in requests[0]["tools"]), requests[0]["tools"]
# Credentials must remain private; preparation fails closed if a # user makes the file readable by other accounts. credentials.chmod(0o644) try: api(route, {"name": "mcp", "enabled": True}) raise AssertionError("world-readable credentials must be rejected") except urllib.error.HTTPError as error: assert error.code == 409, error.read() credentials.chmod(0o600)
# An unavailable server fails the replacement instead of silently # dropping capabilities from a live session. configure([str(root/"missing.py")], startup=3000) try: api(route, {"name": "mcp", "enabled": True}) raise AssertionError("an unavailable MCP server must not enable") except urllib.error.HTTPError as error: assert error.code == 409, error.read() assert not next(item["enabled"] for item in api(route) if item["name"] == "mcp") print("mcp discovery, namespaced calls, env scrubbing, and teardown passed") finally: if connection: with contextlib.suppress(Exception): api("/shutdown", {})
if __name__ == "__main__": server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Provider) threading.Thread(target=server.serve_forever, daemon=True).start() try: run(f"http://127.0.0.1:{server.server_port}/v1") finally: server.shutdown()