Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540"""MCP stdio discovery, credentials, namespaced calls, and teardown through the daemon."""
from typing import Anyfrom concurrent.futures import ThreadPoolExecutor, waitfrom contextlib import closingimport errnoimport jsonimport osimport shutilimport sqlite3import sysimport timeimport unittestfrom pathlib import Path
from harness import Albedo, Provider, exclusive, operation_id, Reply, text, wait_until
SERVER = Path(__file__).with_name("fake_mcp_server.py")
# exclusive: fixture configures global MCP servers, credentials, and daemon environment@exclusiveclass McpTests(unittest.TestCase): def setUp(self): self.message = "ping from albedo" self.python_mode = False self.cell = None
def script(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 self.cell and not answered: return Reply("python", tool_arguments={"code": self.cell}) if mcp_tool and not answered: if self.python_mode: code = ( "import asyncio\n" 'r = await mcp.call("fake/echo", message="ping from python")\n' "print(r.content[0].text)\n" "print(await mcp.tools())\n" "results = await asyncio.gather(\n" ' mcp.call("fake/echo", message="batch one"),\n' ' mcp.call("fake/echo", message="batch two"),\n' " return_exceptions=True,\n" ")\n" "print(results)" ) return Reply("python", tool_arguments={"code": code}) return Reply( "python", tool_name=mcp_tool, tool_arguments={"message": self.message}, ) return text("done")
self.provider = Provider(script) self.addCleanup(self.provider.close)
def prepare(app): app.env.update( ALBEDO_MCP_SECRET="configured-secret", ALBEDO_MCP_CLOSED=str(app.root / "closed"), ALBEDO_MCP_AMBIENT="must-not-reach-the-server", ALBEDO_MCP_CONTROL=str(app.root / "control"), ) (app.root / "control").mkdir() self.configure(app, [str(SERVER)]) app.store_secrets( "mcp", {"fake": {"env": {"FAKE_SECRET": "stored-secret"}}} )
self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session()
def configure(self, app, args, startup=20000, retry=None): path = app.home / "extensions.json" settings: dict[str, Any] = ( json.loads(path.read_text()) if path.exists() else {"models": {"refreshHours": 0}} ) settings["mcp"] = { "servers": { "fake": { "type": "stdio", "command": sys.executable, "args": args, "env": { "FAKE_SECRET": {"env": "ALBEDO_MCP_SECRET"}, "FAKE_MCP_CLOSED": {"env": "ALBEDO_MCP_CLOSED"}, "FAKE_MCP_CONTROL": {"env": "ALBEDO_MCP_CONTROL"}, }, "startupTimeoutMs": startup, } } } if retry is not None: settings["mcp"]["retryMs"] = retry path.write_text(json.dumps(settings))
def tools(self, requests): return [ tool["name"] for tool in requests[0]["tools"] if tool["name"].startswith("mcp_") ]
def mcp(self): with self.app.api(f"/sessions/{self.session}/catalog") as response: catalog = json.load(response) return next( row for row in catalog["discovery"]["candidates"] if row["kind"] == "extension" and row["preference_key"] == "mcp" )
def select_mcp(self, session, enabled): route = f"/sessions/{session}?view=configuration" with self.app.api(route) as response: json.load(response) etag = response.headers["ETag"] self.app.api( route, {"selection": {"extensions": {"mcp": enabled}}}, method="PATCH", headers={"If-Match": etag}, ).close()
def set_mcp(self, enabled): self.select_mcp(self.session, enabled) with self.app.api( f"/sessions/{self.session}/reload", {"target": "session"} ) as response: return json.load(response)
def turn(self, prompt, session=None): session = session or self.session before = len(self.provider.requests) self.app.prompt(session, prompt).close() self.app.idle(session, timeout=60) return [entry["request"] for entry in self.provider.requests[before:]]
@exclusive def test_cold_command_and_recorded_recovery_do_not_block_another_session(self): other = self.app.session() for session, enabled in ((self.session, True), (other, False)): route = f"/sessions/{session}?view=configuration" with self.app.api(route) as response: json.load(response) etag = response.headers["ETag"] self.app.api( route, {"selection": {"extensions": {"mcp": enabled}}}, method="PATCH", headers={"If-Match": etag}, ).close() with self.app.api(f"/sessions/{self.session}?tail=0") as response: self.assertEqual(json.load(response)["kernel"]["state"], "none") marker = self.app.workspace / "cold-mcp-started" gate = self.app.workspace / "cold-mcp-release" os.mkfifo(gate) script = ( "import pathlib; " + f"pathlib.Path({str(marker)!r}).touch(); " + f"open({str(gate)!r}, 'rb').read(1); " + f"exec(pathlib.Path({str(SERVER)!r}).read_text())" ) self.configure(self.app, ["-c", script], startup=30000) input_id = operation_id()
def submit_command(): with self.app.api( f"/sessions/{self.session}/inputs/{input_id}", { "kind": "command", "command_id": "/research", "arguments": {"arguments": "cold composition fixture"}, }, method="PUT", ) as response: return json.load(response)
def release_preparation(): deadline = time.monotonic() + 10 while True: try: descriptor = os.open(gate, os.O_WRONLY | os.O_NONBLOCK) break except OSError as error: if error.errno != errno.ENXIO or time.monotonic() >= deadline: raise time.sleep(0.01) try: os.write(descriptor, b"x") finally: os.close(descriptor)
def complete_independent_turn(): with self.app.api(f"/sessions/{other}?tail=0") as response: self.assertEqual(json.load(response)["id"], other) with self.app.prompt(other, "independent turn") as response: other_input = json.load(response)["id"] self.app.idle(other) with self.app.api(f"/sessions/{other}/inputs/{other_input}") as response: self.assertEqual(json.load(response)["turn"]["state"], "completed")
with ThreadPoolExecutor(max_workers=2) as pool: command = pool.submit(submit_command) deadline = time.monotonic() + 10 while not marker.exists() and time.monotonic() < deadline: time.sleep(0.01) self.assertTrue(marker.exists(), "cold MCP preparation did not start") try: complete_independent_turn() self.assertFalse(command.done(), "command passed the held preparation") finally: release_preparation() self.assertEqual(command.result(timeout=40)["id"], input_id) self.app.idle(self.session) with self.app.api(f"/sessions/{self.session}/inputs/{input_id}") as response: self.assertEqual(json.load(response)["turn"]["state"], "completed") with self.app.api(f"/sessions/{self.session}?tail=0") as response: original_kernel = json.load(response)["kernel"]["instance_id"] self.assertIsNotNone(original_kernel) original_history = self.app.history(self.session)["items"] marker.unlink() self.app.restart() deadline = time.monotonic() + 10 while not marker.exists() and time.monotonic() < deadline: time.sleep(0.01) self.assertTrue(marker.exists(), "recorded kernel preparation did not start") try: complete_independent_turn() finally: release_preparation() deadline = time.monotonic() + 20 while True: with self.app.api(f"/sessions/{self.session}?tail=0") as response: recovered = json.load(response)["kernel"] if recovered["state"] == "attached" or time.monotonic() >= deadline: break time.sleep(0.01) self.assertEqual(recovered["state"], "attached") self.assertEqual(recovered["instance_id"], original_kernel) self.assertEqual(self.app.history(self.session)["items"], original_history)
def test_discovery_namespaced_call_and_scrubbed_credentials(self): self.assertEqual(self.set_mcp(True)["session"]["state"], "applied") self.assertTrue(self.mcp()["effective_enabled"]) requests = self.turn("use the mcp server") self.assertEqual(len(requests), 2) context = requests[0]["instructions"] self.assertNotIn("<extension-context", json.dumps(requests[0]["input"])) self.assertIn('name="mcp"', context) self.assertIn("echo", context) advertised = [ tool["name"] for tool in requests[0]["tools"] if tool["name"].startswith("mcp_") ] self.assertEqual(len(advertised), 1) self.assertTrue(advertised[0].startswith("mcp_fake_echo_")) output = next( item["output"] for item in requests[1]["input"] if item.get("type") == "function_call_output" ) self.assertEqual( json.loads(json.loads(output)["content"][0]["text"]), {"echoed": "ping from albedo", "secret": "stored-secret", "ambient": None}, )
def test_python_binding_calls_and_batches_mcp_operations(self): self.python_mode = True self.set_mcp(True) requests = self.turn("compose mcp calls in python") self.assertEqual(len(requests), 2) output = next( item["output"] for item in requests[1]["input"] if item.get("type") == "function_call_output" ) rendered = json.dumps(output) self.assertIn("ping from python", rendered) self.assertIn("fake", rendered) self.assertIn("echo", rendered) self.assertIn("batch one", rendered) self.assertIn("batch two", rendered)
def test_a_failed_tool_call_is_refused_without_ending_the_turn(self): self.set_mcp(True) self.message = "fail" requests = self.turn("the server will refuse") self.assertEqual(len(requests), 2) output = next( item["output"] for item in requests[1]["input"] if item.get("type") == "function_call_output" ) self.assertIn("MCP request failed", json.loads(output)["error"]) self.assertIn("done", json.dumps(self.app.history(self.session)))
def test_disable_closes_server_and_removes_tools(self): self.set_mcp(True) self.assertEqual(self.set_mcp(False)["session"]["state"], "applied") self.assertFalse(self.mcp()["effective_enabled"]) closed = self.app.root / "closed" deadline = time.monotonic() + 15 while time.monotonic() < deadline and not closed.exists(): time.sleep(0.05) self.assertTrue(closed.exists(), "disabling must close the server") requests = self.turn("the server is gone") self.assertFalse( any(tool["name"].startswith("mcp_") for tool in requests[0]["tools"]) )
def test_insecure_credentials_fail_closed(self): credentials = self.app.home / "creds.json" credentials.chmod(0o644) self.addCleanup(credentials.chmod, 0o600) outcome = self.set_mcp(True) self.assertEqual(outcome["session"]["state"], "failed") self.assertIn("0600", outcome["session"]["failure"]["detail"]) self.assertFalse(self.provider.requests) credentials.chmod(0o600) self.assertEqual(self.set_mcp(True)["session"]["state"], "applied")
def test_an_unavailable_server_is_skipped_and_the_rest_load(self): path = self.app.home / "extensions.json" settings = json.loads(path.read_text()) settings["mcp"]["servers"]["dead"] = { "type": "stdio", "command": str(self.app.root / "missing-server"), "startupTimeoutMs": 3000, } path.write_text(json.dumps(settings)) self.assertEqual(self.set_mcp(True)["session"]["state"], "applied") advertised = self.tools(self.turn("one server is down")) self.assertEqual(len(advertised), 1) self.assertTrue(advertised[0].startswith("mcp_fake_echo_"))
def test_a_server_that_comes_online_joins_the_session_after_a_turn(self): late = self.app.root / "late_server.py" self.configure(self.app, [str(late)], startup=3000, retry=200) self.assertEqual(self.set_mcp(True)["session"]["state"], "applied") self.assertEqual(self.tools(self.turn("nothing is up yet")), []) shutil.copy(SERVER, late) deadline = time.monotonic() + 40 advertised = [] while time.monotonic() < deadline and not advertised: time.sleep(0.5) advertised = self.tools(self.turn("is it up now?")) self.assertEqual(len(advertised), 1, "the server never joined the session") self.assertTrue(advertised[0].startswith("mcp_fake_echo_")) history = json.dumps(self.app.history(self.session)) self.assertIn("capabilities changed", history) self.assertIn("fake", history)
def test_a_known_server_does_not_hold_up_opening_a_session(self): self.set_mcp(True) self.assertEqual(len(self.tools(self.turn("learn the catalogue"))), 1) (self.app.root / "control" / "delay").write_text("3")
session = self.app.session() self.select_mcp(session, True) started = time.monotonic() with self.app.api( f"/sessions/{session}/reload", {"target": "session"} ) as response: self.assertEqual(json.load(response)["session"]["state"], "applied") self.assertLess(time.monotonic() - started, 1.5)
# The saved tools are advertised at once; the call waits for the dial. requests = self.turn("use the mcp server", session) self.assertEqual(len(self.tools(requests)), 1) output = next( item["output"] for item in requests[-1]["input"] if item.get("type") == "function_call_output" ) self.assertIn(self.message, output) self.assertNotIn("capabilities changed", json.dumps(self.app.history(session)))
def test_a_draft_07_tool_with_an_id_property_is_advertised(self): # The TypeScript MCP SDK labels every tool schema draft-07, and tools # often take an `id`. barrel_mcp 4.3.0 read that property name as the # draft-04 `id` keyword and dropped the tool from tools/list. tools = [ { "name": name, "description": name, "inputSchema": { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "required": ["id"], "properties": {"id": {"type": "string"}}, }, } for name in ("stop", "kernels") ] (self.app.root / "control" / "tools").write_text(json.dumps(tools)) self.set_mcp(True) advertised = self.tools(self.turn("list the tools")) self.assertEqual(len(advertised), 2, advertised)
def seen(self): path = self.app.root / "control" / "seen" return path.read_text().split() if path.exists() else []
def output(self, requests): return next( item["output"] for item in requests[-1]["input"] if item.get("type") == "function_call_output" )
def test_a_stateless_server_answers_the_discover_probe(self): (self.app.root / "control" / "mode").write_text("modern") self.assertEqual(self.set_mcp(True)["session"]["state"], "applied") requests = self.turn("use the modern server") self.assertEqual(len(self.tools(requests)), 1) self.assertIn(self.message, self.output(requests)) seen = self.seen() self.assertEqual(seen[0], "server/discover") self.assertNotIn("initialize", seen)
def test_a_server_that_ignores_the_probe_falls_back_quickly(self): # Older servers may ignore a method they do not know before # initialize; the probe waits a second for it, not the startup timeout. (self.app.root / "control" / "mode").write_text("silent") started = time.monotonic() self.assertEqual(self.set_mcp(True)["session"]["state"], "applied") self.assertLess(time.monotonic() - started, 4) self.assertEqual(self.seen()[:2], ["server/discover", "initialize"]) self.assertIn(self.message, self.output(self.turn("use the old server")))
def test_a_server_that_ignores_eof_is_still_stopped(self): control = self.app.root / "control" (control / "mode").write_text("deaf") self.set_mcp(True) pid = int((control / "pid").read_text()) self.set_mcp(False) deadline = time.monotonic() + 15 while True: try: os.kill(pid, 0) except ProcessLookupError: break self.assertLess(time.monotonic(), deadline, "the server outlived close") time.sleep(0.1) self.assertTrue((self.app.root / "closed").exists())
def add_shout(self): schema = { "type": "object", "required": ["message"], "properties": {"message": {"type": "string"}}, } tools = [ {"name": name, "description": name, "inputSchema": schema} for name in ("echo", "shout") ] (self.app.root / "control" / "tools").write_text(json.dumps(tools))
def test_a_changed_catalogue_refreshes_the_session_after_its_turn(self): self.set_mcp(True) self.turn("learn the catalogue") self.add_shout()
session = self.app.session() self.select_mcp(session, True) self.assertEqual( len(self.tools(self.turn("still the saved tools", session))), 1 ) deadline = time.monotonic() + 20 while "capabilities changed" not in json.dumps(self.app.history(session)): self.assertLess(time.monotonic(), deadline, "the session never refreshed") time.sleep(0.1) advertised = self.tools(self.turn("now the new ones", session)) self.assertEqual(len(advertised), 2) self.assertTrue(any(name.startswith("mcp_fake_shout_") for name in advertised))
def test_a_reload_during_a_kernel_release_survives_the_release(self): # A release ends by storing the kernel handle it began with; a reload # applied while its save ran used to be undone by it, and the next # turn advertised the catalogue from before the reload. With no # kernel budget a kernel is released as soon as its turn ends. self.app.restart( prepare=lambda app: app.env.update(ALBEDO_KERNEL_BUDGET_MB="0") ) control = self.app.root / "control" self.set_mcp(True) self.turn("learn the catalogue") # The second session advertises the saved catalogue and checks it in # the background; `hold` keeps that check from seeing the new one. (control / "hold").touch() self.add_shout() session = self.app.session() self.select_mcp(session, True) saving, go = control / "saving", control / "go" # Saving this value, when the kernel is released, waits for `go`. self.cell = ( "import pathlib, time\n" "class Gate:\n" " def __reduce__(self):\n" f" pathlib.Path({str(saving)!r}).touch()\n" f" while not pathlib.Path({str(go)!r}).exists():\n" " time.sleep(0.05)\n" " return (int, ())\n" "gate = Gate()" ) self.assertEqual(len(self.tools(self.turn("hold the release", session))), 1) self.cell = None wait_until(saving.exists, 30, "the kernel was never released") (control / "hold").unlink()
def saved_shout(): database = f"file:{self.app.home / 'albedo.sqlite'}?mode=ro" with closing(sqlite3.connect(database, uri=True)) as db: rows = db.execute("SELECT catalogue FROM mcp_catalogues").fetchall() return any(b"shout" in row[0] for row in rows)
wait_until(saved_shout, 30, "the check never saved the new catalogue") with ThreadPoolExecutor(max_workers=1) as pool: reloaded = pool.submit( self.app.api, f"/sessions/{session}/reload", {"target": "session"} ) # A reload parked behind the release answers only after `go`, so # give it a moment to reach the session before letting the save end. wait([reloaded], timeout=2) go.touch() reloaded.result(timeout=60).close() advertised = self.tools(self.turn("now the new ones", session)) self.assertEqual(len(advertised), 2, advertised)