feat(mcp): skip servers that are offline and add them when they return
A server that could not start used to fail the whole mcp extension for the session, taking every other server's tools with it. It is now left out with a warning, and the rest load. A missing server is probed when the session shows activity, never on a timer, with doubling backoff from the new `retryMs` setting (default 30s). One found up asks the session to refresh at the end of a turn, which leaves the usual "capabilities changed" turn and pins the old prompt until compaction. `extension.Session` gains `refresh(reason)` for this. Saving or enabling a server through settings stays strict, so a mistyped command is still refused where it is typed.
41 unmodified lines42434445464647484950170 unmodified linesExpand all41 unmodified linesA command that tells the agent about a user's change uses the `Note(origin, display, text)` state operation: the note waits in the session's queue, reaches the model at its next step if a run is active or ahead of the user's next message if not, and never starts a turn by itself.
`ManagedPlugin` prepares session-owned contributions together: context, tool instructions, tools, python modules, host routes, and commands. preparation returns a close callback. a failed replacement releases its prepared resources and leaves the old selection active; successful replacement releases the old resources after the swap. use this for connections or an immutable catalog shared by context and tools.
a managed contribution also hears its session through `observe(session, event)`, which runs inside the session actor and so must only send and return. events arrive in order: `CallSent` for each successful turn call as it went out (request, prefix identity, usage, cache marks, profile, endpoint, protocol, timing), `TurnEnded(cancelled)`, `Compacted` after a forced compaction, and `Stirred` for any new activity. the `extension.Session` handle it comes with names the session and offers `call(request, prefix)`: any `types.Request`, sent on the session's upstream as exclusive work that queues submissions behind it, records a provider request row of kind `background` under `prefix`, and never reaches the transcript or the stream. it blocks its caller until the call ends and fails at once while another run holds the session or its kernel is released. `extension.empty()` is the contribution with nothing in it, for record updates; [`warm`](cache-warming.md) is built on these two seams alone.a managed contribution also hears its session through `observe(session, event)`, which runs inside the session actor and so must only send and return. events arrive in order: `CallSent` for each successful turn call as it went out (request, prefix identity, usage, cache marks, profile, endpoint, protocol, timing), `TurnEnded(cancelled)`, `Compacted` after a forced compaction, and `Stirred` for any new activity. the `extension.Session` handle it comes with names the session and offers `call(request, prefix)`: any `types.Request`, sent on the session's upstream as exclusive work that queues submissions behind it, records a provider request row of kind `background` under `prefix`, and never reaches the transcript or the stream. it blocks its caller until the call ends and fails at once while another run holds the session or its kernel is released. `extension.empty()` is the contribution with nothing in it, for record updates; [`warm`](cache-warming.md) is built on these two seams alone. `refresh(reason)` asks the session to prepare its extensions again and records `reason` as a note, with the same capability-changed turn a `/reload` leaves; it answers at once and the session declines while a turn runs, so an extension asks again when `TurnEnded` arrives. [`mcp`](mcp.md) uses it to bring back a server that was offline.
## broken extensions
the daemon starts without an extension it cannot run and says so. `extension.install` quarantines one whose name is blank or already taken, whose requirement is not installed (cascading to what needed it), whose initialiser or schema callback fails or raises, or that the default selection cannot run (a requirement it does not enable, duplicate capabilities, a second compaction strategy). only storage the host owns stops the boot. a quarantined extension is in no session's selection, is listed by `/extensions` with its reason in the `quarantined` field, and any attempt to enable it is refused with that reason.170 unmodified linesExpand all51 unmodified lines525354555656575859606162636465666768697051 unmodified lines## Tools and lifecycle
Discovered capabilities are advertised as `mcp_<server>_<operation>_<hash>`, so two servers that expose the same tool name stay distinct and stable. Resources and prompts, when a server offers them, appear as one `read_resource` and one `get_prompt` operation per server. The catalog and its connections belong to the session composition: enabling, disabling, or reloading the extension prepares a replacement first and closes the previous connections only after a successful swap.
A server that cannot start, initialize, or be discovered fails the whole preparation. The session keeps its previous working composition and reports the unavailable server; capabilities are never silently dropped. Disabling the extension closes every connection and terminates each stdio subprocess.A server that cannot start, initialize, or be discovered is left out of the session and named in a warning; the servers that did connect load as usual. Bad configuration (an invalid name, a missing credential file, a catalogue over the size limit) still fails the whole preparation, and the session keeps its previous working composition. Disabling the extension closes every connection and terminates each stdio subprocess.
### Servers that come back
A server that was left out joins the session once it connects, the way a skill does after a reload: albedo prepares the extension again, appends a "capabilities changed" user turn, and keeps the cached system prompt until compaction.
There is no timer. The extension probes the missing servers when the session shows activity (a submit or the end of a turn), first `retryMs` after the session prepared and then at doubling intervals, capped at twenty times `retryMs`. A server found up waits for the end of a turn and then asks the session to refresh, so it appears after the turn that follows its return, not mid-turn. `retryMs` (default `30000`) sits beside `servers` in the `mcp` section. A server that drops after connecting is not removed; its calls report a transport loss as before.
### Saving a server
The `/mcp` form and the capability toggle stay strict: saving or turning on a server that cannot start is refused (409) and the previous settings and credentials are restored, so a mistyped command is caught where it is typed. A server that is switched off, or off for the session by a capability choice, is not tried.
Tool calls may have side effects, so a failed or interrupted call is never retried automatically. Transport loss is reported as an unknown outcome, which the model must inspect before acting again.
Remote servers are untrusted input. Tool descriptions, resource contents, and prompt text are data, not instructions, and their schemas are bounded before they reach the model.12345678941 unmodified lines5152535454555657585912 unmodified lines7273747576777879808182838485868788899059 unmodified lines150151152153144154155156157158159160151152153154161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194"""MCP stdio discovery, credentials, namespaced calls, and teardown through the daemon."""
from typing import Anyimport jsonimport shutilimport sysimport timeimport unittestimport urllib.error41 unmodified lines self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session() self.route = f"/sessions/{self.session}/extensions"
def configure(self, app, args, startup=20000): 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()12 unmodified lines "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 extension(self, body=None): with self.app.api(self.route, body) as response: return next(item for item in json.load(response) if item["name"] == "mcp")
59 unmodified lines any(tool["name"].startswith("mcp_") for tool in requests[0]["tools"]) )
@exclusive def test_insecure_credentials_and_unavailable_server_fail_closed(self): def test_insecure_credentials_fail_closed(self): credentials = self.app.home / "creds.json" credentials.chmod(0o644) with self.assertRaises(urllib.error.HTTPError) as rejected: self.extension({"name": "mcp", "enabled": True}) self.assertEqual(rejected.exception.code, 409) credentials.chmod(0o600) self.configure(self.app, [str(self.app.root / "missing.py")], startup=3000) with self.assertRaises(urllib.error.HTTPError) as rejected: self.extension({"name": "mcp", "enabled": True}) self.assertEqual(rejected.exception.code, 409) self.assertFalse(self.extension()["enabled"])
@exclusive 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.assertTrue(self.extension({"name": "mcp", "enabled": True})["enabled"]) advertised = self.tools(self.turn("one server is down")) self.assertEqual(len(advertised), 1) self.assertTrue(advertised[0].startswith("mcp_fake_echo_"))
@exclusive 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.assertTrue(self.extension({"name": "mcp", "enabled": True})["enabled"]) 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)221 unmodified linesExpand all22222322422522622622722822923023123223323482 unmodified lines221 unmodified linesExpand all healthy("working"), ]) runtime.observe( session, extension.Session("broken-extension-test", fn(_, _) { Error("no upstream") }), extension.Session( "broken-extension-test", fn(_, _) { Error("no upstream") }, fn(_) { Nil }, ), extension.Stirred, ) answer(host, session, "working") |> should.equal("answered") runtime.stop(host)82 unmodified lines165 unmodified linesExpand all166167168169170171172173174175176519 unmodified linesExpand all6966976986997007017027037047051329 unmodified linesExpand all20352036203720382039204020342035203620372038203920402041204220432041204220432044204520462047204820492050205120522053205420552056349 unmodified linesExpand all165 unmodified linesExpand all /// A background call's worker finished. BackgroundFinished(String, Int, Result(Option(types.Usage), String)) Compact(Option(String), Subject(Result(json.Json, String))) RefreshData(Subject(Result(json.Json, String))) /// An extension asks for a refresh and gives its reason; see /// `extension.Session`. RefreshRequested(String) DrainSteering(String, Subject(Result(List(types.Input), String))) ReadContext(Subject(context_snapshot.Snapshot)) Finished(String, Result(Nil, String)) Down(process.Down)519 unmodified linesExpand all ChangeEffort(level, reply) -> transition(reply, session_provider.change_effort(stirred(state), level))
RefreshData(reply) -> transition(reply, session_extensions.refresh(state)) RefreshRequested(reason) -> actor.continue(session_extensions.refresh_requested(state, reason)) Compact(strategy, reply) -> case turn.running(state.activity), kernel_or_park(state, Compact(strategy, reply))1329 unmodified linesExpand all
/// What an observer may ask of this session. It travels to the observer, so/// it captures the actor's subject, never the state and its transcript.fn background_handle(id: String, self: Session) -> extension.Session { extension.Session( id, extension.Session(id, fn(request, prefix) { case session_run.try_call(self, 600_000, CallInBackground(request, prefix, _)) { Ok(outcome) -> outcome Error(session_run.TimedOut) -> Error("the background call went unanswered for ten minutes") Error(session_run.CalleeDown) -> Error("the session stopped") } }) fn(request, prefix) { case session_run.try_call(self, 600_000, CallInBackground(request, prefix, _)) { Ok(outcome) -> outcome Error(session_run.TimedOut) -> Error("the background call went unanswered for ten minutes") Error(session_run.CalleeDown) -> Error("the session stopped") } }, fn(reason) { process.send(self, RefreshRequested(reason)) }, )}
fn start_queued(state: State) -> State { case turn.running(state.activity) {349 unmodified linesExpand all103 unmodified linesExpand all1041051061071081081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431447 unmodified lines1521531541551301311321331561571581591606 unmodified lines16716816917014814915015117117217317417518 unmodified lines194195196197178198199200201103 unmodified linesExpand all
pub fn refresh( state: session_state.State(message),) -> #(session_state.State(message), Result(json.Json, String)) { refresh_with(state, fn() { refresh_with(state, "session data reloaded from disk", fn() { runtime.refresh_session(state.host, state.info.id) })}
/// Prepares the extensions again because one asked, and says why in a note./// A running turn leaves the session as it is: the extension asks again when/// the turn ends.pub fn refresh_requested( state: session_state.State(message), reason: String,) -> session_state.State(message) { case turn.running(state.activity) { Some(_) -> state None -> case refresh_with(state, reason, fn() { runtime.refresh_session(state.host, state.info.id) }) { #(state, Ok(_)) -> state #(state, Error(error)) -> session_state.emit( state, view.text("error", reason <> ", but the reload failed: " <> error), ) } }}
fn refresh_with( state: session_state.State(message), label: String, reload: fn() -> Result(Option(runtime.Session), String),) -> #(session_state.State(message), Result(json.Json, String)) { case turn.running(state.activity) { Some(_) -> #(state, Error("session must be idle to reload"))7 unmodified lines None -> state } case session_prompt.pin_changed_prompt(state, previous) { Ok(#(state, detail)) -> #( session_state.emit( state, view.text("note", "session data reloaded from disk" <> detail), ), session_state.emit(state, view.text("note", label <> detail)), Ok( json.object([ #("reloaded", json.string("session")), #(6 unmodified lines ]), ), ) Error(error) -> #( session_prompt.reset_prompt_cache( state, "session data reloaded from disk", ), session_prompt.reset_prompt_cache(state, label), Ok( json.object([ #("reloaded", json.string("session")), #(18 unmodified linespub fn save_settings( state: session_state.State(message), change: session_settings.Change,) -> #(session_state.State(message), Result(json.Json, String)) { refresh_with(state, fn() { refresh_with(state, "session data reloaded from disk", fn() { runtime.save_settings(state.host, state.home, state.info.id, change) })}183 unmodified linesExpand all18418518618718818919019119223 unmodified lines21621721821922022122222322422522622722822923023154 unmodified lines183 unmodified linesExpand all home: String, session: String, change: Change, extensions: fn() -> Result(Extensions, String), enabling: fn(String, String) -> Result(Nil, String), after: fn() -> Result(a, String),) -> Result(a, String) { guarded_capability( home,23 unmodified lines case row.shadowed_by, row.valid, change.enabled { Some(_), _, _ -> Error("shadowed candidates cannot be changed") _, False, Some(True) -> Error("invalid candidates cannot be enabled") _, _, Some(True) -> enabling(row.kind, key) |> result.replace(#( row.kind, key, change.scope, change.enabled, )) _, _, _ -> Ok(#(row.kind, key, change.scope, change.enabled)) } } }54 unmodified lines152 unmodified linesExpand all1531541551561571581591601611621631641465 unmodified linesExpand all152 unmodified linesExpand all /// ends and answers its usage; fails at once while another run holds /// the session or its kernel is released. call: fn(types.Request, requests.Prefix) -> Result(Option(types.Usage), String), /// Asks the session to prepare its extensions again, noting `reason` in /// the transcript. Answers at once; the session declines while a turn is /// running, so call it again when the next one ends. refresh: fn(String) -> Nil, )}
pub type Prepared {1465 unmodified linesExpand all18 unmodified lines19202122232324252627282930313233343536373839404130424344454647484950365152535455565758596061626364656667686970717273747576777814 unmodified lines18 unmodified lines after: fn() -> Result(a, String),) -> Result(a, String) { case change { Catalog(change) -> capability_catalog.save(home, session, change, extensions, after) capability_catalog.save( home, session, change, extensions, fn(kind, name) { enabling(session, kind, name) }, after, ) Capability(kind, name, scope, enabled) -> case name != "" && { kind == "skills" || kind == "instructions" || kind == "mcp" } && { scope == "global" || scope == "session" } { True -> { use _ <- result.try(case enabled { option.Some(True) -> enabling(session, kind, name) _ -> Ok(Nil) }) True -> capability(home, session, kind, name, #(scope, enabled), after) capability(home, session, kind, name, #(scope, enabled), after) } False -> Error("choose a capability kind, name, and global or session scope") } MCP(name, server, secrets) -> { use _ <- result.try(mcp.validate_settings(name, server, secrets)) // Session preparation skips a server that is down; saving one refuses // it, so a typo is caught where it is typed. save_mcp(home, name, server, secrets, after) save_mcp(home, name, server, secrets, fn() { use _ <- result.try(case server { option.Some(_) -> mcp.check_connects(session, name, True) option.None -> Ok(Nil) }) after() }) } }}
/// Turning a server on refuses one that cannot start; other kinds have/// nothing to connect to.fn enabling( session: String, kind: String, name: String,) -> Result(Nil, String) { case kind { "mcp" -> mcp.check_connects(session, name, False) _ -> Ok(Nil) }}
@external(erlang, "albedo_settings_store", "capability")fn capability( home: String, session: String,14 unmodified lines123345678910111213141 unmodified line1617181920212223242125262728293031282930313233343233343536373839404142433944454647424348495051525354555657585960616263646566676869707118 unmodified lines909192937394959676979899100101124 unmodified linesExpand all226227228229230231209210232233234212213214215216217218219220235236237238239222223240241242243244245246247248178 unmodified linesExpand all42742842943041043143243343443543643743823 unmodified lines46246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455537 unmodified lines-module(albedo_mcp).
-export([prepare/1, prepare/2, definitions/1, context/1, call/3, close/1, url_allowed/1, validate_settings/3]).-export([prepare/1, prepare/2, definitions/1, context/1, offline/1, observe/3, check/4, call/3, close/1, url_allowed/1, validate_settings/3]).
-define(DEFAULT_STARTUP_MS, 20000).-define(DEFAULT_CALL_MS, 60000).-define(CLOSE_MS, 2500).-define(DEFAULT_RETRY_MS, 30000).-define(RETRY_CEILING, 20).-define(SAFE_ENV, ["HOME", "PATH", "TMPDIR", "TEMP", "TMP", "SystemRoot", "WINDIR"]).
prepare(ConfigJson) -> prepare(ConfigJson, undefined).
1 unmodified line try Config = json:decode(ConfigJson), Servers = maps:get(<<"servers">>, Config, #{}), true = is_map(Servers), Retry = positive_ms(maps:get(<<"retryMs">>, Config, ?DEFAULT_RETRY_MS)), SelectedSession = case Session of undefined -> none; _ -> {some, Session} end, Candidates = lists:sort(maps:to_list(Servers)), Selection = case Candidates of [] -> none; _ -> SelectedSession end, case 'albedo@harness@capabilities':load(albedo_extension_settings:home(), Selection) of {ok, Preferences} -> open_servers(Candidates, [], Preferences); {ok, Preferences} -> open_servers(Candidates, Preferences, Retry); Error -> Error end catch _:_ -> {error, <<"MCP configuration is invalid">>} end.
open_servers([], Opened, _) -> Servers = lists:reverse(Opened), case catalogue(Servers) of {ok, Ops, Ctx} -> {ok, #{servers => Servers, operations => Ops, context => Ctx}}; {error, Reason} -> close_servers(Servers), {error, Reason} end;open_servers([{Name, Config} | Rest], Opened, Preferences) ->open_servers(Candidates, Preferences, Retry) -> open_servers(Candidates, [], [], Preferences, Retry).
%% A server that cannot start is left out and named; one that is badly%% configured still fails the whole preparation.open_servers([], Opened, Offline, _, Retry) -> finish(lists:reverse(Opened), lists:reverse(Offline), Retry);open_servers([{Name, Config} | Rest], Opened, Offline, Preferences, Retry) -> case {maps:get(<<"enabled">>, Config, true), 'albedo@harness@capabilities':enabled(Preferences, <<"mcp">>, Name)} of {true, {ok, true}} -> case open_server(Name, Config) of {ok, Server} -> open_servers(Rest, [Server | Opened], Preferences); {ok, Server} -> open_servers(Rest, [Server | Opened], Offline, Preferences, Retry); {unavailable, _} -> open_servers(Rest, Opened, [{Name, Config} | Offline], Preferences, Retry); {error, Reason} -> close_servers(Opened), {error, Reason} end; {false, _} -> open_servers(Rest, Opened, Preferences); {_, {ok, false}} -> open_servers(Rest, Opened, Preferences); {false, _} -> open_servers(Rest, Opened, Offline, Preferences, Retry); {_, {ok, false}} -> open_servers(Rest, Opened, Offline, Preferences, Retry); {_, {error, Reason}} -> close_servers(Opened), {error, Reason} end.
finish(Servers, Offline0, Retry) -> {Live, Offline1, Operations, Context} = catalogue(Servers, Offline0), Offline = lists:keysort(1, Offline1), Names = [maps:get(advertised, O) || O <- Operations], case length(Operations) =< 512 andalso length(Names) =:= length(lists:usort(Names)) of true -> Watcher = watch(Offline, Retry), {ok, #{servers => Live, operations => Operations, offline => [N || {N, _} <- Offline], watcher => Watcher, context => iolist_to_binary(lists:join("\n", Context))}}; false -> close_servers(Live), {error, <<"MCP catalogue is too large or ambiguous">>} end.
open_server(Name, Config) when is_binary(Name), is_map(Config) -> case valid_name(Name) of false -> {error, <<"MCP server name is invalid">>}; true ->18 unmodified lines }), {ok, Pid} = barrel_mcp_client:start(Spec), case await_ready(Pid, Startup) of ok -> {ok, #{name => Name, pid => Pid, call_timeout => Call, config => Config}}; {error, _} -> close_client(Pid), {error, unavailable(Name)} {error, _} -> close_client(Pid), {unavailable, Name} end catch _:_ -> {error, unavailable(Name)} _:_ -> {unavailable, Name} end.
transport_spec(#{<<"type">> := <<"http">>, <<"url">> := Url} = Config, Secrets, Spec) when is_binary(Url), byte_size(Url) > 0 ->124 unmodified linesExpand all catch _:Caught -> {error, Caught} end end.
%% A server whose tools cannot be listed is closed and counted offline with%% the ones that never started.catalogue(Servers) -> catalogue(Servers, [], []).catalogue(Servers, Offline) -> catalogue(Servers, Offline, [], [], []).
catalogue([], Operations0, Context) -> Operations = lists:reverse(Operations0), Names = [maps:get(advertised, O) || O <- Operations], case length(Operations) =< 512 andalso length(Names) =:= length(lists:usort(Names)) of true -> {ok, Operations, iolist_to_binary(lists:join("", lists:reverse(Context)))}; false -> {error, <<"MCP catalogue is too large or ambiguous">>} end;catalogue([#{name := Name} = Server | Rest], Operations, Context) ->catalogue([], Offline, Live, Operations, Context) -> {lists:reverse(Live), lists:reverse(Offline), lists:reverse(Operations), lists:reverse(Context)};catalogue([#{name := Name, config := Config} = Server | Rest], Offline, Live, Operations, Context) -> case discover(Server) of {ok, Ops, Text} -> catalogue(Rest, lists:reverse(Ops) ++ Operations, [Text | Context]); {error, _} -> {error, unavailable(Name)} {ok, Ops, Text} -> catalogue(Rest, Offline, [Server | Live], lists:reverse(Ops) ++ Operations, [Text | Context]); {error, _} -> close_servers([Server]), catalogue(Rest, [{Name, Config} | Offline], Live, Operations, Context) end.
discover(#{name := Name, pid := Pid, call_timeout := Timeout, config := Config}) -> Enabled = maps:get(<<"enabledTools">>, Config, null),178 unmodified linesExpand all case lists:search(fun(Item) -> maps:get(Key, Item) =:= Name end, Items) of {value, Item} -> Item; false -> erlang:error(not_found) end.
close(#{servers := Servers}) -> close_servers(Servers), nil;close(#{servers := Servers, watcher := Watcher}) -> stop_watch(Watcher), close_servers(Servers), nil;close(_) -> nil.
close_servers(Servers) -> lists:foreach(fun(S) -> close_client(maps:get(pid, S)) end, Servers).
23 unmodified linesoptional_binary(_) -> erlang:error(invalid_value).
unavailable(Name) -> <<"MCP server '", Name/binary, "' is unavailable">>.
%% The servers that were configured and enabled but could not be reached.offline(#{offline := Offline}) -> Offline.
%% Connects to one saved server and hangs up, so the settings form can refuse a%% server that cannot start. A server that is switched off, or off for `Session`%% when `Preferences` is true, is not tried.check(ConfigJson, Session, Name, Preferences) -> try Servers = maps:get(<<"servers">>, json:decode(ConfigJson), #{}), {ok, Prefs} = 'albedo@harness@capabilities':load(albedo_extension_settings:home(), {some, Session}), Wanted = Preferences =:= false orelse 'albedo@harness@capabilities':enabled(Prefs, <<"mcp">>, Name) =:= {ok, true}, case maps:find(Name, Servers) of {ok, #{<<"enabled">> := false}} -> {ok, nil}; {ok, Config} when Wanted -> case probe(Name, Config) of true -> {ok, nil}; false -> {error, unavailable(Name)} end; _ -> {ok, nil} end catch _:_ -> {error, <<"MCP configuration is invalid">>} end.
probe(Name, Config) -> case open_server(Name, Config) of {ok, Server} -> close_servers([Server]), true; _ -> false end.
%% Tells the handle's watcher what its session just did. `Refresh` takes the%% reason a server is back and asks the session to prepare its tools again.observe(#{watcher := Watcher}, TurnEnded, Refresh) when is_pid(Watcher) -> Watcher ! {event, TurnEnded, Refresh}, nil;observe(_, _, _) -> nil.
%% Servers that were offline at preparation are probed when the session shows%% activity, never on a timer, so an idle daemon does not poll a dead host.%% Probes back off after each failure; one that finds a server up waits for a%% turn to end and then asks for the refresh, repeating until the session%% takes it (which closes this handle and with it the watcher).watch([], _) -> none;watch(Offline, Retry) -> spawn(fun() -> watch(Offline, Retry, [], 0, now_ms() + Retry) end).
watch(Offline, Retry, Back, Failures, Next) -> receive stop -> ok; {event, TurnEnded, Refresh} -> Ended = folded_events(TurnEnded), {Back1, Failures1, Next1} = maybe_probe(Offline, Retry, Back, Failures, Next), case Back1 =/= [] andalso Ended of true -> Refresh(<<"MCP server ", (hd(Back1))/binary, " is reachable again">>); false -> ok end, watch(Offline, Retry, Back1, Failures1, Next1) end.
%% Folds the events already queued into one, so a burst costs one probe; the%% turn counts as ended if any of them ended it.folded_events(TurnEnded) -> receive {event, Later, _} -> folded_events(TurnEnded orelse Later) after 0 -> TurnEnded end.
maybe_probe(_, _, Back, Failures, Next) when Back =/= [] -> {Back, Failures, Next};maybe_probe(Offline, Retry, [], Failures, Next) -> Now = now_ms(), case Now >= Next of false -> {[], Failures, Next}; true -> case [Name || {Name, Config} <- Offline, probe(Name, Config)] of [] -> Wait = min(Retry bsl Failures, Retry * ?RETRY_CEILING), {[], Failures + 1, now_ms() + Wait}; Back -> {Back, 0, now_ms()} end end.
stop_watch(Watcher) when is_pid(Watcher) -> Watcher ! stop, ok;stop_watch(_) -> ok.
now_ms() -> erlang:monotonic_time(millisecond).
%% Validate persisted transport references and secret patches without resolving%% environment variables or starting a client.validate_settings(Name, Server, SecretsJSON) -> try37 unmodified lines39 unmodified lines40414243444445464748495051525354555456575859606162636465636667686970112 unmodified linesExpand all18318418518618718818919019119219319419519619719819920020120220320420 unmodified lines22522622722822923023123223349 unmodified lines2832842852862872882892902912922932942952962972982993003013023033043053063073082 unmodified lines31131231331431531631731831932032132232332432532632732832933033118 unmodified lines39 unmodified lines )}
type Config { Config(servers: Dict(String, Server)) Config(servers: Dict(String, Server), retry_ms: Int)}
type Handle
type Definition { Definition(name: String, description: String, parameters: Json)}
const default_retry_ms = 30_000
fn default_config() -> Config { Config(dict.new()) Config(dict.new(), default_retry_ms)}
fn config_decoder() -> decode.Decoder(Config) { use servers <- decode.optional_field( "servers", dict.new(), decode.dict(decode.string, server_decoder()), ) use retry_ms <- decode.optional_field("retryMs", default_retry_ms, decode.int) decode.success(Config(servers)) decode.success(Config(servers, retry_ms))}
fn server_decoder() -> decode.Decoder(Server) { use kind <- decode.field("type", decode.string)112 unmodified linesExpand all }, fn(_) { None }, ) }), warnings: list.map(native_offline(handle), fn(name) { "MCP server '" <> name <> "' is unavailable; its tools join the session once it connects" }), observe: fn(session: extension.Session, event) { case event { extension.Stirred -> native_observe(handle, False, session.refresh) extension.TurnEnded(_) -> native_observe(handle, True, session.refresh) _ -> Nil } }, close: fn() { native_close(handle) }, ), ) }20 unmodified lines |> dict.to_list |> list.map(fn(entry) { #(entry.0, encode_server(entry.1)) }), ), ), #("retryMs", json.int(config.retry_ms)), ]) |> json.to_string}
49 unmodified lines
@external(erlang, "albedo_mcp", "context")fn native_context(handle: Handle) -> String
@external(erlang, "albedo_mcp", "offline")fn native_offline(handle: Handle) -> List(String)
@external(erlang, "albedo_mcp", "observe")fn native_observe( handle: Handle, turn_ended: Bool, refresh: fn(String) -> Nil,) -> Nil
@external(erlang, "albedo_mcp", "check")fn native_check( config: String, session: String, name: String, preferences: Bool,) -> Result(Nil, String)
@external(erlang, "albedo_mcp", "call")fn native_call( handle: Handle, name: String,2 unmodified lines
@external(erlang, "albedo_mcp", "close")fn native_close(handle: Handle) -> Nil
/// Connects to the saved server `name` and hangs up. A server that is not/// saved or is switched off passes, and so does one that `session`'s/// capability choices leave off when `preferences` is true; one that cannot/// start is an error.pub fn check_connects( session: String, name: String, preferences: Bool,) -> Result(Nil, String) { use config <- result.try(load_config()) native_check(encode_config(config), session, name, preferences)}
/// Validate a persisted server even when MCP is disabled in the session.pub fn validate_settings( name: String, server: Option(String),18 unmodified lines