"""MCP stdio server -- a thin, capability-only facade over the broker unix socket. Design rule: this process contains NO crypto and holds NO secret material. It only translates MCP tool calls into broker socket operations. By construction there is no get_password / get_secret / reveal / dump tool -- the tools below map to broker adapters that return sanitized metadata only. """ from __future__ import annotations import json import sys from . import __version__ from .errors import BrokerError from .server import call PROTOCOL_VERSION = "2024-11-05" SERVER_NAME = "agent-broker" def _str(name, desc, required=True): p = {"type": "string", "description": desc} return p, (required and [name] or []) def _obj(properties, required): return {"type": "object", "properties": properties, "required": required, "additionalProperties": False} TOOLS = [ { "name": "vault_status", "description": "Broker status: locked/unlocked, configured, ref count. No secrets.", "inputSchema": _obj({}, []), "_method": ("status", lambda a: {}), }, { "name": "vault_lock", "description": "Lock the broker immediately (drops all in-memory credentials).", "inputSchema": _obj({}, []), "_method": ("lock", lambda a: {}), }, { "name": "vault_list_refs", "description": "List credential references and safe metadata (never secret values).", "inputSchema": _obj({}, []), "_method": ("list_refs", lambda a: {}), }, { "name": "vault_find_for_url", "description": "List credential refs whose policy allows the given URL.", "inputSchema": _obj({"url": {"type": "string"}}, ["url"]), "_method": ("find_for_url", lambda a: {"url": a["url"]}), }, { "name": "credential_http_request", "description": "Run an HTTP request with a credential attached by the broker " "(bearer/basic per item policy). Returns sanitized response.", "inputSchema": _obj({ "credential_ref": {"type": "string"}, "method": {"type": "string", "enum": ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]}, "url": {"type": "string"}, "headers": {"type": "object"}, "body": {"type": "string"}, "follow_redirects": {"type": "boolean"}, "ticket": {"type": "string", "description": "confirmation ticket from a CONFIRMATION_REQUIRED response"}, }, ["credential_ref", "method", "url"]), "_method": ("http_request", lambda a: a), }, { "name": "ssh_exec", "description": "Run a single command over SSH using the credential's host policy. " "Private keys/passwords never leave the broker.", "inputSchema": _obj({ "credential_ref": {"type": "string"}, "host": {"type": "string"}, "command": {"type": "string"}, "port": {"type": "integer"}, "ticket": {"type": "string"}, }, ["credential_ref", "host", "command"]), "_method": ("ssh_exec", lambda a: a), }, { "name": "git_operation", "description": "git clone/fetch/push using vault credentials via askpass. " "The credential never appears in the remote URL or argv.", "inputSchema": _obj({ "credential_ref": {"type": "string"}, "op": {"type": "string", "enum": ["clone", "fetch", "push"]}, "repository": {"type": "string"}, "workdir": {"type": "string"}, "dest": {"type": "string"}, "remote": {"type": "string"}, "refspec": {"type": "string"}, "branch": {"type": "string"}, "ticket": {"type": "string"}, }, ["credential_ref", "op", "repository"]), "_method": ("git_op", lambda a: a), }, { "name": "cli_run", "description": "Run an allowlisted CLI profile with the credential injected by " "the broker (env/stdin/tmpfs-file). No arbitrary commands.", "inputSchema": _obj({ "profile": {"type": "string"}, "credential_ref": {"type": "string"}, "args": {"type": "array", "items": {"type": "string"}}, "ticket": {"type": "string"}, }, ["profile", "credential_ref"]), "_method": ("cli_run", lambda a: a), }, { "name": "browser_login", "description": "Login to a site via agent-browser using the broker " "credential provider plugin. Secrets flow plugin->agent-browser, " "never through the agent.", "inputSchema": _obj({ "credential_ref": {"type": "string"}, "url": {"type": "string"}, "session": {"type": "string"}, "username_selector": {"type": "string"}, "password_selector": {"type": "string"}, "submit_selector": {"type": "string"}, "ticket": {"type": "string"}, }, ["credential_ref"]), "_method": ("browser_login", lambda a: a), }, { "name": "browser_fill_credential", "description": "Fill username+password on the CURRENT page (no navigation) using " "the agent-browser credential provider and the item's allowed origin.", "inputSchema": _obj({ "credential_ref": {"type": "string"}, "url": {"type": "string"}, "session": {"type": "string"}, "username_selector": {"type": "string"}, "password_selector": {"type": "string"}, "submit_selector": {"type": "string"}, "ticket": {"type": "string"}, }, ["credential_ref"]), "_method": ("browser_fill_credential", lambda a: a), }, ] # Hard guard: even a bug in the table above cannot create a secret- # retrieval tool -- the names are statically checked at startup. _FORBIDDEN = ("password", "secret", "token", "private_key", "reveal", "dump", "get_") for _t in TOOLS: n = _t["name"].lower() assert not any(b in n for b in _FORBIDDEN), f"forbidden tool name {n}" del _FORBIDDEN, _t, n def _result(obj): return {"content": [{"type": "text", "text": json.dumps(obj, indent=2, sort_keys=True)}]} def _error_result(e: BrokerError): out = {"error": e.to_dict()} if getattr(e, "ticket", None): out["ticket"] = e.ticket return {"content": [{"type": "text", "text": json.dumps(out)}], "isError": True} def _handle(req: dict) -> dict | None: method = req.get("method") rid = req.get("id") if method == "initialize": return {"jsonrpc": "2.0", "id": rid, "result": { "protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {}}, "serverInfo": {"name": SERVER_NAME, "version": __version__}, }} if method == "notifications/initialized" or method == "notifications/cancelled": return None if method == "ping": return {"jsonrpc": "2.0", "id": rid, "result": {}} if method == "tools/list": tools = [{k: v for k, v in t.items() if not k.startswith("_")} for t in TOOLS] return {"jsonrpc": "2.0", "id": rid, "result": {"tools": tools}} if method == "tools/call": params = req.get("params") or {} name = params.get("name") args = params.get("arguments") or {} tool = next((t for t in TOOLS if t["name"] == name), None) if tool is None: return {"jsonrpc": "2.0", "id": rid, "result": _error_result(BrokerError("DENIED", f"unknown tool {name!r}"))} socket_method, mapper = tool["_method"] try: res = call(socket_method, mapper(args)) return {"jsonrpc": "2.0", "id": rid, "result": _result(res)} except BrokerError as e: return {"jsonrpc": "2.0", "id": rid, "result": _error_result(e)} if rid is not None: return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": "method not found"}} return None def main(): if sys.stdin.isatty(): print("agent-broker-mcp is the MCP stdio server — not a CLI.", file=sys.stderr) print("For interactive commands use: agent-broker ", file=sys.stderr) return 2 while True: line = sys.stdin.readline() if not line: return 0 line = line.strip() if not line: continue try: req = json.loads(line) except json.JSONDecodeError: continue resp = _handle(req) if resp is not None: sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() if __name__ == "__main__": sys.exit(main())