Route agents across providers and align Claude OAuth MCP requests
22 unmodified lines232425262728293022 unmodified linesmist = ">= 6.0.3 and < 7.0.0"gleam_http = ">= 4.4.0 and < 5.0.0"yamerl = ">= 0.10.0 and < 1.0.0"barrel_mcp = ">= 4.0.0 and < 5.0.0"gleam_crypto = ">= 1.0.0 and < 2.0.0"
[dev_dependencies]gleeunit = ">= 1.0.0 and < 2.0.0"39 unmodified lines4041424344454647487 unmodified lines39 unmodified lines]
[requirements]barrel_mcp = { version = ">= 4.0.0 and < 5.0.0" }gleam_crypto = { version = ">= 1.0.0 and < 2.0.0" }gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" }gleam_http = { version = ">= 4.4.0 and < 5.0.0" }gleam_json = { version = ">= 3.1.0 and < 4.0.0" }gleam_otp = { version = ">= 1.3.0 and < 2.0.0" }7 unmodified lines45 unmodified lines4647484950515253543 unmodified lines58596061626364656619 unmodified lines86878889908889919293949596979899132 unmodified linesExpand all23223323423523023623723823924024124224324424524624724824925025125225325425525625719 unmodified lines45 unmodified lines
class Provider(http.server.BaseHTTPRequestHandler): """Answers every request in plain text naming what it was asked.""" requests = [] paths = [] lock = threading.Lock()
def log_message(self, *_): pass3 unmodified lines latest_user = next((content_text(item) for item in reversed(request["messages"]) if item.get("role") == "user"), "") with self.lock: self.requests.append(latest_user) self.paths.append(self.path) answer = "answered: " + latest_user.splitlines()[-2 if "</mail>" in latest_user else -1] self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.end_headers()19 unmodified lines workspace.mkdir() (home / "extensions.json").write_text(json.dumps({"models": {"refreshHours": 0}})) (home / "config.json").write_text(json.dumps({ "active": "alpha", "providers": { "providers": {"alpha": {"baseUrl": endpoint + "/alpha/v1", "apiKey": "key", "model": "fixture-alpha", "protocol": "chat_completions"}}, "alpha": {"baseUrl": endpoint + "/alpha/v1", "apiKey": "key", "model": "fixture-alpha", "protocol": "chat_completions"}, "beta": {"baseUrl": endpoint + "/beta/v1", "apiKey": "key", "model": "fixture-beta", "protocol": "chat_completions"}, }, })) env = dict(os.environ, HOME=str(Path(directory) / "user-home"), ALBEDO_HOME=str(home), ALBEDO_PARENT_PID=str(os.getpid())) if TOOLCHAIN and shutil.which("gleam") is None:132 unmodified linesExpand all deleted = api(f"/sessions/{lead}?tree=1", method="DELETE") assert deleted["deleted"] == 2, deleted listed = [s["id"] for s in api("/sessions")] assert lead not in listed and coder not in listed, listed print("spawn, forwarded answers, mail by name and id, and parent deletion order hold")
# A child can use a different configured provider without changing its parent. try: api(f"/sessions/{radio}/children", {"name": "missing", "task": "should not run", "model": "beta/no-such-model"}) raise AssertionError("accepted an unavailable qualified model") except urllib.error.HTTPError as error: assert error.code == 409, error.code for name, model in (("explicit", "beta/fixture-beta"), ("inferred", "fixture-beta")): child = api(f"/sessions/{radio}/children", {"name": name, "task": "cross-provider task", "model": model}) assert child["session"]["provider"] == "beta", child assert child["session"]["model"] == "fixture-beta", child assert child["member"]["parent"] == radio, child settle(child["session"]["id"]) assert sum("/beta/" in path for path in Provider.paths) >= 2, Provider.paths print("spawn, cross-provider routing, forwarded answers, mail, and deletion hold") finally: if connection: with contextlib.suppress(Exception): api("/shutdown", {})19 unmodified lines42 unmodified lines43444546474849505170 unmodified lines12212312412512612512612712812913013113213313413584 unmodified lines42 unmodified lines
LEAD = """m = await agents.models()assert "beta/fixture-beta" in m, mkid = await agents.self.spawn("count to three", name="scout", model=m[0])try: await agents.self.spawn("again", name="scout2", model="")except TypeError as e:70 unmodified lines workspace.mkdir() (home / "extensions.json").write_text(json.dumps({"models": {"refreshHours": 0}})) (home / "config.json").write_text(json.dumps({ "active": "alpha", "providers": { "providers": {"alpha": {"baseUrl": endpoint + "/alpha/v1", "apiKey": "key", "model": "fixture-alpha", "protocol": "chat_completions"}}, "alpha": {"baseUrl": endpoint + "/alpha/v1", "apiKey": "key", "model": "fixture-alpha", "protocol": "chat_completions"}, "beta": {"baseUrl": endpoint + "/beta/v1", "apiKey": "key", "model": "fixture-beta", "protocol": "chat_completions"}, }, })) env = dict(os.environ, HOME=str(Path(directory) / "user-home"), ALBEDO_HOME=str(home), ALBEDO_PARENT_PID=str(os.getpid())) if TOOLCHAIN and shutil.which("gleam") is None:84 unmodified lines82 unmodified lines83848586878889908788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515642 unmodified lines19920020120214720320420520620720820921021121221360 unmodified lines27427527627721621727827928028128228328428528628728828929029135 unmodified lines3273283293302613313323333343353363373383393403415 unmodified lines34734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444540 unmodified lines82 unmodified lines None, types.defaults, ) let assert Ok(openai_api.Exchange(headers: headers, body: body, ..)) = wire.encode("sk-ant-oat-test", request) assert list_key(headers, "anthropic-beta") == Ok("claude-code-20250219,oauth-2025-04-20") assert list_key(headers, "user-agent") == Ok("claude-cli/2.1.283") wire.encode( "sk-ant-oat-test", "11111111-2222-4333-8444-555555555555", string.repeat("a", 64), "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", request, ) let assert Ok(betas) = list_key(headers, "anthropic-beta") assert string.split(betas, ",") == [ "claude-code-20250219", "oauth-2025-04-20", "interleaved-thinking-2025-05-14", "redact-thinking-2026-02-12", "thinking-token-count-2026-05-13", "context-management-2025-06-27", "prompt-caching-scope-2026-01-05", "mid-conversation-system-2026-04-07", "per-turn-control-2026-07-01", "mid-conversation-tool-changes-2026-07-01", "extended-cache-ttl-2025-04-11", ] assert list_key(headers, "user-agent") == Ok("claude-cli/2.1.283 (external, cli)") assert list_key(headers, "x-stainless-lang") == Ok("js") assert list_key(headers, "x-stainless-runtime") == Ok("node") assert list_key(headers, "x-stainless-package-version") == Ok("0.112.1") assert list_key(headers, "x-stainless-retry-count") == Ok("0") assert list_key(headers, "x-stainless-timeout") == Ok("600") assert list_key(headers, "x-stainless-arch") == Ok("arm64") assert list_key(headers, "x-stainless-os") == Ok("MacOS") assert list_key(headers, "x-stainless-runtime-version") == Ok("v26.3.0") assert list_key(headers, "x-claude-code-session-id") == Ok("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee") let assert Ok(value) = json.parse(string_tree.to_string(body), decode.dynamic) let assert Ok(system) = decode.run( value, decode.at(["system"], decode.list(decode.at(["text"], decode.string))), ) let assert Ok(user_id) = decode.run(value, decode.at(["metadata", "user_id"], decode.string)) let assert Ok(identity_fields) = json.parse(user_id, decode.dynamic) let assert Ok("11111111-2222-4333-8444-555555555555") = decode.run(identity_fields, decode.at(["account_uuid"], decode.string)) let assert Ok("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee") = decode.run(identity_fields, decode.at(["session_id"], decode.string)) let assert Ok(device) = decode.run(identity_fields, decode.at(["device_id"], decode.string)) assert device == string.repeat("a", 64) let assert Ok(["all"]) = decode.run( value, decode.at( ["context_management", "edits"], decode.list(decode.at(["keep"], decode.string)), ), ) let assert Ok(ttls) = decode.run( value, decode.at( ["system"], decode.list( decode.one_of(decode.at(["cache_control", "ttl"], decode.string), or: [ decode.success("none"), ]), ), ), ) assert ttls == ["none", "1h", "1h"] let assert [billing, identity, local] = system assert identity == "You are Claude Code, Anthropic's official CLI for Claude." assert local == "local instructions" assert string.starts_with(42 unmodified lines None, types.defaults, ) let assert Ok(openai_api.Exchange(body: body, ..)) = wire.encode("token", request) wire.encode( "token", "11111111-2222-4333-8444-555555555555", string.repeat("a", 64), "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", request, ) let assert Ok(value) = json.parse(string_tree.to_string(body), decode.dynamic) let assert Ok([tool]) = decode.run(value, decode.at(["tools"], decode.list(decode.dynamic))) let assert Ok("object") =60 unmodified lines [], None, options, ) let assert Ok(openai_api.Exchange(body: body, ..)) = wire.encode("token", request) let assert Ok(openai_api.Exchange(headers: headers, body: body, ..)) = wire.encode( "token", "11111111-2222-4333-8444-555555555555", string.repeat("a", 64), "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", request, ) let assert Ok(betas) = list_key(headers, "anthropic-beta") assert string.contains(betas, "effort-2025-11-24") let assert Ok(value) = json.parse(string_tree.to_string(body), decode.dynamic) let assert Ok("adaptive") = decode.run(value, decode.at(["thinking", "type"], decode.string)) let assert Ok("max") =35 unmodified lines None, types.defaults, ) let assert Ok(openai_api.Exchange(body: body, ..)) = wire.encode("token", request) wire.encode( "token", "11111111-2222-4333-8444-555555555555", string.repeat("a", 64), "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", request, ) let assert Ok(value) = json.parse(string_tree.to_string(body), decode.dynamic) let assert Ok(names) = decode.run( value,5 unmodified lines assert string.contains(encoded, "tool_use") assert string.contains(encoded, "hello")}
pub fn claude_mcp_tools_use_subscription_namespace_and_round_trip_test() { let name = "mcp_web_extract_web_search_9cdda7e075" let alias = "mcp__albedo__web_extract_web_search_9_be5f1c469f2fb0ae" assert wire.claude_name(name) == alias let long = "mcp_" <> string.repeat("a", 49) <> "_0123456789" assert string.length(wire.claude_name(long)) <= 64 assert wire.claude_name(long) != wire.claude_name(long <> "b") let tool = types.Tool( name, "Search", json.object([#("type", json.string("object"))]), False, ) let request = types.Request( "claude-opus-5-5", None, [types.User("search")], [tool], None, types.defaults, ) let assert Ok(openai_api.Exchange(body: body, ..)) = wire.encode( "token", "11111111-2222-4333-8444-555555555555", string.repeat("a", 64), "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", request, ) let assert Ok(value) = json.parse(string_tree.to_string(body), decode.dynamic) let assert Ok([declared]) = decode.run( value, decode.at(["tools"], decode.list(decode.at(["name"], decode.string))), ) assert declared == alias let chunks = [ "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"usage\":{\"input_tokens\":12,\"output_tokens\":0}}}", "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"tool_1\",\"name\":\"" <> alias <> "\",\"input\":{}}}", "{\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"}}", "{\"type\":\"message_stop\"}", ] let assert Ok(turn) = feed_all(stream.reducer("claude-opus-5-5", [tool]), chunks) let assert [call] = turn.tool_calls assert call.name == name let replay = types.Request( "claude-opus-5-5", None, [ types.User("search"), types.Replay(list_first(turn.output)), types.ToolOutput("tool_1", "ok", []), ], [tool], None, types.defaults, ) let assert Ok(openai_api.Exchange(body: replay_body, ..)) = wire.encode( "token", "11111111-2222-4333-8444-555555555555", string.repeat("a", 64), "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", replay, ) let assert Ok(replayed) = json.parse(string_tree.to_string(replay_body), decode.dynamic) let assert Ok(names) = decode.run( replayed, decode.at( ["messages"], decode.list(decode.at( ["content"], decode.list( decode.one_of(decode.at(["name"], decode.string), or: [ decode.success(""), ]), ), )), ), ) assert list.any(names, fn(items) { list.contains(items, alias) })}
fn feed_all( reducer: reducer.Reducer, chunks: List(String),) -> Result(types.Turn, types.Error) {40 unmodified lines793 unmodified linesExpand all7947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718727 unmodified lines88088188288388488588688788888989089189289389489589689789889990090190282082182282390390490590690790890991097 unmodified lines100810091010101192993093193293393493593610121013101410151016101710181019102010211022102310241025102610271028102910301006 unmodified linesExpand all793 unmodified linesExpand all } }}
fn provider_models( host: runtime.Runtime, provider: configuration.Provider,) -> List(String) { list.unique([ provider.model, ..runtime.model_names(host, provider.extension, "") ])}
fn child_model( state: State, parent: conversation.Info, requested: String,) -> Result(#(configuration.Provider, String), String) { use profiles <- result.try(configuration.providers(state.config.home)) let models = list.map(profiles, fn(profile) { #(profile, provider_models(state.host, profile)) }) let qualified = list.find_map(models, fn(pair) { let #(profile, available) = pair let prefix = profile.name <> "/" case string.starts_with(requested, prefix) { True -> Ok(#( profile, string.drop_start(requested, string.length(prefix)), available, )) False -> Error(Nil) } }) case qualified { Ok(#(profile, model, available)) -> case list.contains(available, model) { True -> Ok(#(profile, model)) False -> Error("model is not available from provider " <> profile.name) } Error(_) -> { use current <- result.try(configuration.named( state.config.home, parent.provider, )) let model = case requested { "" -> parent.model _ -> requested } case list.contains(provider_models(state.host, current), model) { True -> Ok(#(current, model)) False -> { let matches = list.filter(models, fn(pair) { let #(profile, available) = pair profile.name != current.name && list.contains(available, model) }) case matches { [#(profile, _)] -> Ok(#(profile, model)) [] -> Ok(#(current, model)) _ -> Error( "model is available from multiple providers; use provider/model", ) } } } } }}
/// A child of `parent`, linked, running, and handed its task.fn create_child( state: State, parent: String,7 unmodified lines dict.get(state.sessions, parent) |> result.replace_error("session not found"), ) use _ <- result.try(family.valid_name(name)) use #(provider, selected_model) <- result.try(child_model( state, above, model, )) let efforts = session_provider.model_efforts( state.host, state.config.home, provider.name, selected_model, ) let info = conversation.Info( ..above, id: new_id(), title: name, stage: conversation.Idle, last_assistant_at: None, model: case model { "" -> above.model _ -> model }, provider: provider.name, model: selected_model, protocol: provider.protocol, effort: extension.default_effort(efforts), ) use _ <- result.try(conversation.create(db, info)) case family.link(db, info.id, parent, name) { Ok(member) -> Ok(#(info, member))97 unmodified lines agents.Models(id) -> { let listed = case dict.get(state.sessions, id) { Error(_) -> [] Ok(#(info, _)) -> { let catalog = case configuration.named(state.config.home, info.provider) { Ok(provider) -> runtime.model_names(state.host, provider.extension, "") Error(_) -> [] } list.unique([info.model, ..catalog]) let profiles = configuration.providers(state.config.home) |> result.unwrap([]) let same = profiles |> list.filter(fn(profile) { profile.name == info.provider }) |> list.flat_map(provider_models(state.host, _)) let other = profiles |> list.filter(fn(profile) { profile.name != info.provider }) |> list.flat_map(fn(profile) { provider_models(state.host, profile) |> list.map(fn(model) { profile.name <> "/" <> model }) }) list.unique([info.model, ..list.append(same, other)]) } } #(state, Ok(json.array(listed, json.string))) }1006 unmodified linesExpand all14 unmodified lines15161718191920212223183 unmodified linesExpand all14 unmodified linesimport gleam/option.{type Option, None, Some}import gleam/resultimport gleam/string
const instructions = "Agents are sessions you spawn to work in parallel; every call is async. agents.self is your handle (id, name, depth, parent). Call await agents.models() once before spawning, then child = await agents.self.spawn(task, name=\"scout\", model=<one of models()>, deliverable=None, evidence_bar=None, falsifier=None): it returns as soon as the child exists, never with its answer. Answers arrive later as <mail> in your conversation and start your next turn, so spawn independent children back to back and end your turn instead of waiting; never sleep or poll. Talk with await mail.submit(child, text). await agents.self.children() and siblings() return live snapshots (running, closed). await child.cancel() stops its turn; await child.close() when you are done with it keeps its transcript and files and frees its kernel. Only the user deletes agents: if one should go, ask them. Children nest at most 3 deep and 12 open per parent; mail any session by id instead of nesting to reach it. Children share your workspace, so give two children the same files only if one only reads."const instructions = "Agents are sessions you spawn to work in parallel; every call is async. agents.self is your handle (id, name, depth, parent). Call await agents.models() once before spawning; other providers appear as provider/model, and selecting one routes the child through that provider. Then child = await agents.self.spawn(task, name=\"scout\", model=<one of models()>, deliverable=None, evidence_bar=None, falsifier=None): it returns as soon as the child exists, never with its answer. Answers arrive later as <mail> in your conversation and start your next turn, so spawn independent children back to back and end your turn instead of waiting; never sleep or poll. Talk with await mail.submit(child, text). await agents.self.children() and siblings() return live snapshots (running, closed). await child.cancel() stops its turn; await child.close() when you are done with it keeps its transcript and files and frees its kernel. Only the user deletes agents: if one should go, ask them. Children nest at most 3 deep and 12 open per parent; mail any session by id instead of nesting to reach it. Children share your workspace, so give two children the same files only if one only reads."
pub fn extension() -> extension.Extension { extension.Extension( "agents",183 unmodified linesExpand all1233456761 unmodified lines6970717273747576777879808182838485868788899091929394959697989910010110269 unmodified lines-module(albedo_claude_auth).
-export([exchange/4, account/1, access/2, expire/2]).-export([exchange/4, account/1, access/2, profile/2, expire/2]).
-define(KEY, <<"anthropic">>).-define(CLIENT_ID, <<"9d1c250a-e61b-44d9-88ed-5944d1962f5e">>).-define(TOKEN_URL, "https://platform.claude.com/v1/oauth/token").61 unmodified lines first_access(Ordered, Path, Session); _ -> {error, <<"Claude is not authenticated; run /login and add a Claude account">>} end.
%% A session-stable device and UUID must accompany the OAuth account identity.%% Resolve the account from the token, never from a claimed client header.profile(Access, Session) -> _ = application:ensure_all_started(inets), _ = application:ensure_all_started(ssl), Request = {"https://api.anthropic.com/api/oauth/profile", [{"authorization", "Bearer " ++ unicode:characters_to_list(Access)}, {"accept", "application/json"}]}, Options = [{timeout, ?HTTP_TIMEOUT_MS}, {connect_timeout, 10000}, {ssl, albedo_credentials:tls_options("api.anthropic.com")}], case httpc:request(get, Request, Options, [{body_format, binary}]) of {ok, {{_, 200, _}, _, Body}} -> try json:decode(Body) of #{<<"account">> := #{<<"uuid">> := UUID}} when is_binary(UUID), byte_size(UUID) =:= 36 -> Device = binary:encode_hex(crypto:hash(sha256, <<"albedo:claude-device:", UUID/binary>>), lowercase), {ok, {UUID, Device, session_uuid(Session)}}; _ -> {error, <<"Claude profile has no account UUID">>} catch _:_ -> {error, <<"Claude profile response is invalid">>} end; _ -> {error, <<"could not read Claude account profile">>} end.
session_uuid(Session) -> <<A:32, B:16, _:4, C:12, _:2, D:14, E:48, _/binary>> = crypto:hash(sha256, Session), iolist_to_binary(io_lib:format("~8.16.0b-~4.16.0b-4~3.16.0b-~4.16.0b-~12.16.0b", [A, B, C, 16#8000 bor D, E])).
credentials(Data) -> case maps:get(?KEY, Data, []) of List when is_list(List) -> [V || V <- List, is_map(V), maps:get(<<"type">>, V, <<>>) =:= <<"oauth">>]; V when is_map(V) -> case maps:get(<<"type">>, V, <<>>) of <<"oauth">> -> [V]; _ -> [] end;69 unmodified lines79 unmodified lines80818283848485868788899091929394959697988889909192939495969798991001011021031041059910010110210310410510610710810911011111211311411511611711811910710810912012112212312412558 unmodified lines18418518618718818919019119219319419579 unmodified lines "claude" -> case context.protocol { types.ChatCompletions -> Some({ use access <- result.map(native_access( use access <- result.try(native_access( context.home, context.session, )) use #(account, device, session) <- result.try(native_profile( access, context.session, )) Ok( extension.Upstream( endpoint, types.ChatCompletions, fn(request, on_event) { use exchange <- result.try(wire.encode( access, extension.Upstream( endpoint, types.ChatCompletions, fn(request, on_event) { use exchange <- result.try(wire.encode(access, request)) openai_api.exchange( exchange, stream.reducer(request.model, request.tools), on_event, ) }, fn(error) { case error { types.HttpError(401, _) -> { native_expire(context.home, access) Some( "Claude rejected this access token; send your message again to refresh it, or run /login", ) account, device, session, request, )) openai_api.exchange( exchange, stream.reducer(request.model, request.tools), on_event, ) }, fn(error) { case error { types.HttpError(401, _) -> { native_expire(context.home, access) Some( "Claude rejected this access token; send your message again to refresh it, or run /login", ) } _ -> None } _ -> None } }, }, ), ) }) _ -> Some(Error("Claude provider requires the chat_completions protocol"))58 unmodified lines
@external(erlang, "albedo_claude_auth", "expire")fn native_expire(home: String, access: String) -> Nil
@external(erlang, "albedo_claude_auth", "profile")fn native_profile( access: String, session: String,) -> Result(#(String, String, String), String)
@external(erlang, "albedo_claude_auth", "access")fn native_access(home: String, session: String) -> Result(String, String)325 unmodified linesExpand all32632732832933033033133233333422 unmodified lines325 unmodified linesExpand all
fn original_name(name: String, tools: List(types.Tool)) -> String { tools |> list.find(fn(tool) { string.lowercase(tool.name) == string.lowercase(name) string.lowercase(wire.claude_name(tool.name)) == string.lowercase(name) }) |> result.map(fn(tool) { tool.name }) |> result.unwrap(name)}22 unmodified lines