diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index cd7e676d1..b7f50ba97 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -462,6 +462,31 @@ print(env.ready, env.problems) # False ['shop: seeded'] asked, or answers no `WebsiteReadiness`, is a problem itself. A card or child without `ready`, an MCP server's or an older site's, isn't counted, so an env with no site serving it is ready with no problems. +## Websites with tools + +A website env may also be an MCP server, so an agent can act on a site through tools as well as its pages. Give a +`WebsiteEnvironment` `@tool` (or `@resource`) methods, as an MCP env has; they need the `mcp` package. The site then +serves them at `/mcp` (streamable HTTP) beside its pages, `@api` routes and data plane, and its card declares that +endpoint as its `mcp` interface (`additionalInterfaces`) and lists the tools under `capabilities.tools`. A site without +tools serves no `/mcp` and declares no interface; `serves_mcp()` says which a site is. + +```python +@environment_card(name="shop") +@website(domains=["shop.example"], static="dist") +class Shop(WebsiteEnvironment): + @tool() + def place_order(self, item: str, ctx: ToolContext) -> str: + """Place an order for item, as the caller.""" + ... +``` + +A tool runs in the site's process, so it changes what its pages show and what `data/get` returns, and it reads the +universe's time (`self.now()`) as a page's request does. The gateway reads each site's card when it starts, and serves +the tools of a site whose card declares an `mcp` interface as it does an MCP env's: listed to agents with the rest, +called with the caller's role in `ToolContext`. The site's `/mcp` is one of the paths its card advertises, which its +browsing route (`/website//`) never reaches. `python -m agentenv_protocol.website.conformance` checks a declared +endpoint (`mcp`): it answers `initialize` and lists the tools the card does. + ## Computer drivers A computer driver is what runs inside a computer an agent uses through screenshots and input. Write diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py b/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py index 22f1bd43b..a22850aa2 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py @@ -7,8 +7,11 @@ the card's ``website/v1`` ``ready``, whose answer must be a ``WebsiteReadiness`` (by the declared health path, deprecated, for a card without it); the data plane's and card's paths not the site's pages; and the data plane the card advertises, whose round trip at the card's ``url`` (data/get, data/reset, data/get, which no longer returns the -exported data, data/add of what the first data/get returned, data/get again) returns what it was given. A failed one is -``failed``. The recommended checks are what an SDK site does and the gateway uses when a server offers it: +exported data, data/add of what the first data/get returned, data/get again) returns what it was given; and, for a +card declaring an ``mcp`` interface, a site that is an MCP server too, that endpoint answering ``initialize`` and +listing, every page of ``tools/list`` read, no tool its card doesn't (streamable HTTP, ``MCP_PROTOCOL_VERSION``); a +card's tool it doesn't list is warned, since a hidden-by-default tool is listed to no caller until one enables it. +A failed one is ``failed``. The recommended checks are what an SDK site does and the gateway uses when a server offers it: ``X-Universe-Now`` on every answer the other checks got, ``clock/v1``'s ``sync_time`` against a clock the checker serves, the forwarded headers wherever an answer shows them (on ``/`` and on a GET route with its trailing slash toggled, which a server may redirect), and the cookies those answers set: none tied to a domain, the session path's the one the card names, and @@ -38,13 +41,15 @@ from pydantic import ValidationError from ..client import find_extension, find_extension_method -from ..types import METHOD_ADD, METHOD_GET, METHOD_RESET, RPC_PATH, WELL_KNOWN_PATH +from ..types import MCP_TRANSPORT, METHOD_ADD, METHOD_GET, METHOD_RESET, RPC_PATH, WELL_KNOWN_PATH from .clock import CLOCK_EXTENSION_URI, UNIVERSE_NOW_HEADER, iso, parse_time from .readiness import DEFAULT_HEALTH_PATH, WEBSITE_EXTENSION_URI, WEBSITE_READY_OP, WebsiteReadiness Status = Literal["passed", "failed", "skipped", "warned"] CLOCK_START = datetime.datetime(2031, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc) CLOCK_SLACK_S = 120 +MCP_PROTOCOL_VERSION = "2025-06-18" +MCP_LIST_PAGES = 100 @dataclasses.dataclass(frozen=True) @@ -112,6 +117,7 @@ def run(self) -> list[Result]: steps = (("card", True, lambda: self._card(response, card, site)), ("any_host", True, self._any_host), ("health", True, lambda: self._health(card, health)), ("universe_now", False, self._universe_now), ("reserved_paths", True, self._reserved_paths), ("data_plane", True, lambda: self._data_plane(card)), + ("mcp", True, lambda: self._mcp(card)), ("clock", False, lambda: self._clock(card)), ("forwarded", False, lambda: self._forwarded(site)), ("cookies", False, lambda: self._cookies(site))) outcomes = {name: _outcome(step) for name, _required, step in steps if name != "universe_now"} @@ -228,6 +234,55 @@ def _data_plane(self, card: dict) -> tuple[Status, str]: return "failed", "data/get after data/reset and data/add of its own export returned other data" return "passed", f"{len(exported)} part(s) round-tripped through reset and add" + def _mcp(self, card: dict) -> tuple[Status, str]: + """The MCP endpoint the card declares answers ``initialize`` and lists no tool the card doesn't, reading up to + ``MCP_LIST_PAGES`` pages of ``tools/list``; its answers needn't carry ``X-Universe-Now``, which the MCP + transport doesn't promise.""" + interfaces = [i for i in card.get("additionalInterfaces") or [] if isinstance(i, dict) + and i.get("transport") == MCP_TRANSPORT] + if not interfaces: + return "skipped", "the card declares no mcp interface" + path = interfaces[0].get("url") + if not isinstance(path, str) or not path: + return "failed", f"the card's mcp interface has no url: {path!r}" + url = self._url(path) + headers = {"Accept": "application/json, text/event-stream"} + first = self.client.post(url, headers=headers, json={ + "jsonrpc": "2.0", "id": "initialize", "method": "initialize", + "params": {"protocolVersion": MCP_PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": "agentenv-website-conformance", "version": "1"}}}) + server = _mcp_result(first, "initialize", "initialize").get("serverInfo") or {} + if session := first.headers.get("mcp-session-id"): + headers["Mcp-Session-Id"] = session + try: + self.client.post(url, headers=headers, json={"jsonrpc": "2.0", "method": "notifications/initialized"}) + served: list = [] + cursor = None + for _page in range(MCP_LIST_PAGES): + listed = _mcp_result(self.client.post(url, headers=headers, json={ + "jsonrpc": "2.0", "id": "tools/list", "method": "tools/list", + **({"params": {"cursor": cursor}} if cursor else {})}), "tools/list", "tools/list") + if not isinstance(listed, dict) or not isinstance(listed.get("tools"), list): + return "failed", f"tools/list at {path} answered no tools list: {str(listed)[:200]}" + served += [t.get("name") for t in listed["tools"] if isinstance(t, dict)] + cursor = listed.get("nextCursor") + if not cursor: + break + else: + return "failed", f"tools/list at {path} still had a next page after {MCP_LIST_PAGES}" + finally: + if session: + with contextlib.suppress(httpx.HTTPError): + self.client.delete(url, headers=headers) + advertised = {t.get("name") for t in (card.get("capabilities") or {}).get("tools") or [] if isinstance(t, dict)} + if unadvertised := sorted(set(served) - advertised): + return "failed", f"tools/list at {path} named {unadvertised}, which the card doesn't list" + name = server.get("name") or "the server" + if unlisted := sorted(advertised - set(served)): + return "warned", (f"{name} at {path} lists {len(served)} of the card's tools, not {unlisted}: hidden by " + "default, or missing") + return "passed", f"{name} at {path} lists the card's {len(served)} tool(s)" + def _clock(self, card: dict) -> tuple[Status, str]: if find_extension(card, CLOCK_EXTENSION_URI) is None: return "failed", f"the card declares no {CLOCK_EXTENSION_URI}" @@ -310,6 +365,48 @@ def _outcome(step: Callable[[], tuple[Status, str]]) -> tuple[Status, str]: return "failed", f"{type(e).__name__}: {e}" +def _mcp_result(response: httpx.Response, method: str, request_id: str) -> dict: + """The result of the JSON-RPC reply to request ``request_id``: the answer's JSON, or the reply among an event + stream's messages, each event's data lines joined, as ``ToolSession`` reads them.""" + if response.headers.get("content-type", "").startswith("text/event-stream"): + messages = [] + for data in _sse_data(response.text): + try: + messages.append(json.loads(data)) + except ValueError: + raise RuntimeError(f"{method} answered an event that isn't JSON: {data[:200]!r}") from None + else: + try: + messages = [json.loads(response.text)] + except ValueError: + raise RuntimeError(f"{method} answered {response.status_code} with no JSON-RPC answer: " + f"{response.text[:200]!r}") from None + body = next((m for m in messages if isinstance(m, dict) and m.get("id") == request_id + and ("result" in m or "error" in m)), None) + if body is None or "result" not in body: + raise RuntimeError(f"{method} answered {response.status_code}: {str(body or messages)[:200]}") + return body["result"] or {} + + +def _sse_data(text: str) -> Iterator[str]: + """The data of each ``message`` event in a server-sent event stream, its data lines joined; an event the stream + ends inside is dropped, as are events with no data.""" + event, data = "message", [] + for line in text.splitlines(): + if not line: + payload = "\n".join(data) + if payload.strip() and event == "message": + yield payload + event, data = "message", [] + continue + field, _, value = line.partition(":") + value = value[1:] if value.startswith(" ") else value + if field == "data": + data.append(value) + elif field == "event": + event = value + + def _field_error(error: dict) -> str: """One of a ``ValidationError``'s errors as a person reads it: where, then what.""" where = ".".join(map(str, error.get("loc") or ())) or "the answer" diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py b/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py index ba7acadee..89296f618 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py @@ -8,6 +8,10 @@ (``@reset_data``, ``@add_data``, ``@get_data``) as any AgentEnvEnvironment does: ``data/get`` is how agentenv reads what a task left a site with. +A site may also be an MCP server, as an MCP env is: its ``@tool`` and ``@resource`` methods are served at ``/mcp`` +(streamable HTTP, of the ``mcp`` package), which its card declares as its ``mcp`` interface and lists the tools of, so +an agent can act on the site through tools beside its pages. A site with none serves no ``/mcp``. + One route is built in, replaced by a site's own ``@api`` route at its very path and answering before one whose path parameters would also match it: ``/api/session``, which a frontend calls to learn who is signed in, signing the acting user in when nobody is (GET), or to sign out (DELETE).""" @@ -32,9 +36,12 @@ from ..agent_env_environment import ( AgentEnvEnvironment, + _RESOURCE_ATTR, + _TOOL_ATTR, AgentEnvStarletteApplication, _marked_members, _maybe_await, + create_fastmcp_app, extension, ) from .clock import CLOCK_EXTENSION_URI, UNIVERSE_NOW_HEADER, ClockUnavailable, UniverseClock, clock_patch_script, iso @@ -164,8 +171,15 @@ def create_app(self) -> Starlette: api_routes = self.routes() taken = {openapi_path(r.path) for r in api_routes} root = self._site_path(declaration.static) if declaration.static is not None else None - asgi = Starlette(middleware=[Middleware(_Forwarded), Middleware(_Universe, env=self)]) - AgentEnvStarletteApplication(self._build_card(), self).add_routes_to_app(asgi) + middleware = [Middleware(_Forwarded), Middleware(_Universe, env=self)] + if self.serves_mcp(): + self.mcp = create_fastmcp_app(self, card=self._build_card()) + served = self.mcp.streamable_http_app() + asgi = Starlette(middleware=middleware, lifespan=served.router.lifespan_context) + asgi.router.routes.extend(served.routes) + else: + asgi = Starlette(middleware=middleware) + AgentEnvStarletteApplication(self._build_card(), self).add_routes_to_app(asgi) if SESSION_PATH not in taken: asgi.router.routes.append(Route(SESSION_PATH, self._session, methods=["GET", "DELETE"])) asgi.router.routes.extend(Route(r.path, _unreserved(endpoint(getattr(self, r.fn.__name__), r, self.context)), @@ -175,6 +189,11 @@ def create_app(self) -> Starlette: self._mounted = True return asgi + def serves_mcp(self) -> bool: + """Whether the site is an MCP server too: it has ``@tool`` or ``@resource`` methods, or forwards tools.""" + return any(True for _ in _marked_members(self, _TOOL_ATTR)) or any( + True for _ in _marked_members(self, _RESOURCE_ATTR)) or bool(self.forwarded_tools()) + def mount(self, app: Any) -> Any: raise TypeError(f"{type(self).__name__} builds its own app with create_app(); set app = ... to serve more") diff --git a/packages/agentenv-protocol/tests/test_website_conformance.py b/packages/agentenv-protocol/tests/test_website_conformance.py index caca4765e..6d2ce401f 100644 --- a/packages/agentenv-protocol/tests/test_website_conformance.py +++ b/packages/agentenv-protocol/tests/test_website_conformance.py @@ -20,7 +20,7 @@ from agentenv_protocol.website import conformance from agentenv_protocol.website.clock import UNIVERSE_NOW_HEADER, iso -CHECKS = ["card", "any_host", "health", "universe_now", "reserved_paths", "data_plane", "clock", "forwarded", "cookies"] +CHECKS = ["card", "any_host", "health", "universe_now", "reserved_paths", "data_plane", "mcp", "clock", "forwarded", "cookies"] def statuses(results: list[conformance.Result]) -> dict[str, str]: @@ -68,7 +68,7 @@ def sdk_site(notes: WebsiteEnvironment) -> TestClient: def test_an_sdk_site_passes_every_check(sdk_site: TestClient) -> None: results = conformance.check("http://testserver", client=sdk_site) assert [r.name for r in results] == CHECKS - assert all(r.status == "passed" for r in results), results + assert all(r.status == ("skipped" if r.name == "mcp" else "passed") for r in results), results assert sdk_site.post("/agentenv", json={"jsonrpc": "2.0", "id": 1, "method": "data/get"}).json()["result"][ "parts"][0]["data"] == {"notes": [{"id": 1, "text": "hello"}]} @@ -173,7 +173,7 @@ async def unrouted(request: Request, exc: Exception) -> Response: def test_a_handwritten_server_meeting_the_contract_passes() -> None: results = conformance.check("http://testserver", client=TestClient(_handwritten())) - assert statuses(results) == dict.fromkeys(CHECKS, "passed"), results + assert statuses(results) == {**dict.fromkeys(CHECKS, "passed"), "mcp": "skipped"}, results @pytest.mark.parametrize(("answer", "status", "detail"), [ @@ -317,9 +317,9 @@ def test_a_server_with_no_card_health_or_header_fails_the_contract_and_warns_of_ bare = Starlette(routes=[Route("/", lambda request: JSONResponse({"not": "html"}))]) results = conformance.check("http://testserver", client=TestClient(bare)) assert statuses(results) == {"card": "failed", "any_host": "failed", "health": "failed", "universe_now": "warned", - "reserved_paths": "passed", "data_plane": "failed", "clock": "warned", - "forwarded": "passed", "cookies": "skipped"} - assert [r.name for r in results if r.required] == ["card", "any_host", "health", "reserved_paths", "data_plane"] + "reserved_paths": "passed", "data_plane": "failed", "mcp": "skipped", + "clock": "warned", "forwarded": "passed", "cookies": "skipped"} + assert [r.name for r in results if r.required] == ["card", "any_host", "health", "reserved_paths", "data_plane", "mcp"] def test_a_page_and_a_health_path_with_no_card_or_data_plane_fail_the_contract() -> None: diff --git a/packages/agentenv-protocol/tests/test_website_mcp.py b/packages/agentenv-protocol/tests/test_website_mcp.py new file mode 100644 index 000000000..05ed7fd07 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_website_mcp.py @@ -0,0 +1,234 @@ +"""A website env that is an MCP server too: its @tool methods served at /mcp, declared on its card, beside its pages, +API and data plane; a site without tools serving no /mcp; and conformance checking the endpoint a card declares.""" +from __future__ import annotations + +import json +from pathlib import Path + +import httpx +import pytest +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse, Response +from starlette.routing import Route +from starlette.testclient import TestClient + +from agentenv_protocol import DataPart, ToolContext, add_data, environment_card, get_data, reset_data, tool +from agentenv_protocol.website import UNIVERSE_NOW_HEADER, WebsiteEnvironment, api, website +from agentenv_protocol.website import conformance + +MCP_HEADERS = {"accept": "application/json, text/event-stream", "content-type": "application/json"} + + +def _shop(*, tools: bool = True, static: Path | None = None) -> WebsiteEnvironment: + if static is not None: + (static / "index.html").write_text("shop") + declared = website(domains=["shop.test"], static=None if static is None else str(static)) + + class Shop(WebsiteEnvironment): + def __init__(self) -> None: + super().__init__() + self.orders: list[dict] = [] + + @api.get("/api/orders", name="list_orders") + def list_orders(self) -> list: + return self.orders + + @reset_data + def reset(self) -> None: + self.orders = [] + + @add_data + def add(self, parts: list) -> None: + for part in parts: + self.orders += part.data.get("orders", []) + + @get_data + def export(self) -> list: + return [DataPart(data={"orders": self.orders})] + + if not tools: + return environment_card(name="shop")(declared(Shop))() + + class ToolShop(Shop): + @tool() + def place_order(self, item: str, ctx: ToolContext) -> str: + """Place an order for item, as the caller.""" + self.orders.append({"item": item, "by": ctx.role, "at": self.now().isoformat()}) + return f"order {len(self.orders)} placed" + + return environment_card(name="shop")(declared(ToolShop))() + + +def _answer(response: httpx.Response) -> dict: + text = response.text + if response.headers.get("content-type", "").startswith("text/event-stream"): + text = [line[len("data:"):] for line in text.splitlines() if line.startswith("data:")][-1] + return json.loads(text) + + +def _mcp(client: TestClient) -> dict: + """Headers of an initialized MCP session with the site.""" + first = client.post("/mcp", headers=MCP_HEADERS, json={ + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}}) + assert first.status_code == 200, first.text + headers = {**MCP_HEADERS, "mcp-session-id": first.headers["mcp-session-id"]} + assert client.post("/mcp", headers=headers, json={"jsonrpc": "2.0", "method": "notifications/initialized"}).is_success + return headers + + +def test_a_site_with_tools_declares_its_mcp_endpoint_and_tools_on_its_card() -> None: + with TestClient(_shop().create_app()) as client: + card = client.get("/.well-known/agent-env.json").json() + assert card["additionalInterfaces"] == [{"url": "/mcp", "transport": "mcp"}] + assert [t["name"] for t in card["capabilities"]["tools"]] == ["place_order"] + assert "ctx" not in card["capabilities"]["tools"][0]["inputSchema"]["properties"] + assert {e["uri"] for e in card["capabilities"]["extensions"]} >= {"urn:agentenv:website/v1", "urn:agentenv:clock/v1"} + + +def test_a_tool_call_acts_on_the_site_its_pages_and_data_plane_read() -> None: + site = _shop() + with TestClient(site.create_app()) as client: + headers = _mcp(client) + listed = _answer(client.post("/mcp", headers=headers, json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"})) + assert [t["name"] for t in listed["result"]["tools"]] == ["place_order"] + called = client.post("/mcp", headers={**headers, "AgentEnv-Role": "world"}, json={ + "jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "place_order", "arguments": {"item": "milk"}}}) + assert _answer(called)["result"]["content"][0]["text"] == "order 1 placed" + assert UNIVERSE_NOW_HEADER in called.headers + assert [o["item"] for o in client.get("/api/orders").json()] == ["milk"] + data = client.post("/agentenv", json={"jsonrpc": "2.0", "id": 1, "method": "data/get"}).json() + assert data["result"]["parts"][0]["data"]["orders"][0]["item"] == "milk" + assert site.orders[0]["by"] == "world" + + +def test_a_site_without_tools_serves_no_mcp_endpoint() -> None: + site = _shop(tools=False) + assert not site.serves_mcp() + with TestClient(site.create_app()) as client: + card = client.get("/.well-known/agent-env.json").json() + answer = client.post("/mcp", headers=MCP_HEADERS, json={"jsonrpc": "2.0", "id": 1, "method": "initialize"}) + assert card["additionalInterfaces"] == [] + assert "tools" not in (card["capabilities"] or {}) or not card["capabilities"]["tools"] + assert answer.status_code == 404 + assert site.mcp is None + + +def test_a_site_with_tools_passes_conformance_with_its_mcp_endpoint_checked(tmp_path: Path) -> None: + with TestClient(_shop(static=tmp_path).create_app()) as client: + results = {r.name: r for r in conformance.check("http://testserver", client=client)} + assert results["mcp"].status == "passed", results["mcp"] + assert "1 tool(s)" in results["mcp"].detail + assert all(r.passed for r in results.values()), results + + +def test_a_site_without_tools_skips_the_mcp_check(tmp_path: Path) -> None: + with TestClient(_shop(tools=False, static=tmp_path).create_app()) as client: + results = {r.name: r for r in conformance.check("http://testserver", client=client)} + assert results["mcp"].status == "skipped" + + +def _declaring(mcp: Route | None) -> Starlette: + """A server whose card declares an MCP interface listing the tool ``a``, answering at /mcp as ``mcp`` does.""" + card = {"name": "x", "additionalInterfaces": [{"url": "/mcp", "transport": "mcp"}], + "capabilities": {"tools": [{"name": "a", "inputSchema": {"type": "object"}}]}} + routes = [Route("/.well-known/agent-env.json", lambda request: JSONResponse(card))] + return Starlette(routes=routes + ([mcp] if mcp is not None else [])) + + +def _listing(*pages: tuple[str, ...]) -> Route: + """An MCP endpoint whose tools/list answers ``pages`` in turn, each pointing at the next by its cursor.""" + async def answer(request: Request) -> Response: + body = await request.json() + if body.get("method") == "initialize": + return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": {"serverInfo": {"name": "x"}}}) + if body.get("method") == "tools/list": + at = int((body.get("params") or {}).get("cursor") or 0) + result = {"tools": [{"name": n} for n in pages[at]], **({"nextCursor": str(at + 1)} if at + 1 < len(pages) + else {})} + return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": result}) + return Response(status_code=202) + + return Route("/mcp", answer, methods=["POST"]) + + +@pytest.mark.parametrize(("mcp", "status", "detail"), [ + (_listing(("a",)), "passed", "lists the card's 1 tool(s)"), + (_listing((), ("a",)), "passed", "lists the card's 1 tool(s)"), + (_listing(("a", "b")), "failed", "named ['b'], which the card doesn't list"), + (_listing(()), "warned", "lists 0 of the card's tools, not ['a']: hidden by default, or missing"), + (None, "failed", "initialize answered 404"), +]) +def test_the_mcp_check_reads_the_endpoint_the_card_declares(mcp: Route | None, status: str, detail: str) -> None: + client = TestClient(_declaring(mcp)) + card = client.get("/.well-known/agent-env.json").json() + found = conformance._outcome(lambda: conformance._Checker("http://testserver", client, "127.0.0.1")._mcp(card)) + assert found[0] == status and detail in found[1], found + + +def test_a_hidden_by_default_tool_on_the_card_warns_without_failing_the_run(tmp_path: Path) -> None: + site = _shop(static=tmp_path) + + class Gated(type(site)): + tool_access = True + + @tool(hidden_by_default=True) + def refund(self, item: str) -> str: + """Refund an order.""" + return "refunded" + + gated = environment_card(name="shop")(Gated)() + with TestClient(gated.create_app()) as client: + results = {r.name: r for r in conformance.check("http://testserver", client=client)} + assert results["mcp"].status == "warned" and "['refund']" in results["mcp"].detail + assert all(r.passed for r in results.values()), results + + +@pytest.mark.parametrize("result", [None, {}, {"tools": "a"}, {"tools": None}]) +def test_a_tools_list_answering_no_list_of_tools_fails_the_check(result: object) -> None: + async def answer(request: Request) -> Response: + body = await request.json() + if body.get("method") == "initialize": + return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": {"serverInfo": {"name": "x"}}}) + if body.get("method") == "tools/list": + return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": result}) + return Response(status_code=202) + + client = TestClient(_declaring(Route("/mcp", answer, methods=["POST"]))) + card = client.get("/.well-known/agent-env.json").json() + found = conformance._outcome(lambda: conformance._Checker("http://testserver", client, "127.0.0.1")._mcp(card)) + assert found[0] == "failed" and "answered no tools list" in found[1], found + + +def _streaming(*events: str) -> Route: + """An MCP endpoint answering each request with an event stream of ``events``, ``{id}`` the request's id.""" + async def answer(request: Request) -> Response: + body = await request.json() + if "id" not in body: + return Response(status_code=202) + stream = "".join(event.replace("{id}", json.dumps(body["id"])) for event in events) + return Response(stream, media_type="text/event-stream") + + return Route("/mcp", answer, methods=["POST"]) + + +def test_the_mcp_check_reads_whole_events_and_the_reply_to_its_own_request() -> None: + reply = '{"jsonrpc": "2.0", "id": {id},\n"result": {"serverInfo": {"name": "x"}, "tools": [{"name": "a"}]}}' + mcp = _streaming('event: message\ndata: {"jsonrpc": "2.0", "method": "notifications/message", "params": {}}\n\n', + ": a comment\n\n", + "".join(f"data: {line}\n" for line in reply.split("\n")) + "\n", + 'data: {"jsonrpc": "2.0", "method": "notifications/progress", "params": {}}\n\n') + client = TestClient(_declaring(mcp)) + card = client.get("/.well-known/agent-env.json").json() + found = conformance._outcome(lambda: conformance._Checker("http://testserver", client, "127.0.0.1")._mcp(card)) + assert found == ("passed", "x at /mcp lists the card's 1 tool(s)") + + +def test_an_mcp_interface_on_another_server_fails_the_check_with_nothing_sent_there() -> None: + sent = [] + client = httpx.Client(transport=httpx.MockTransport(lambda request: sent.append(request) or httpx.Response(500))) + card = {"additionalInterfaces": [{"url": "http://elsewhere.test/mcp", "transport": "mcp"}]} + found = conformance._outcome(lambda: conformance._Checker("http://site.test", client, "127.0.0.1")._mcp(card)) + assert found[0] == "failed" and sent == [] diff --git a/packages/agentenv-protocol/tests/test_website_notes_example.py b/packages/agentenv-protocol/tests/test_website_notes_example.py index 6af42c236..0c58f4542 100644 --- a/packages/agentenv-protocol/tests/test_website_notes_example.py +++ b/packages/agentenv-protocol/tests/test_website_notes_example.py @@ -111,9 +111,9 @@ def ready(client: TestClient) -> dict: def test_the_example_passes_every_conformance_check(client: TestClient) -> None: results = conformance.check("http://testserver", client=client) - assert {r.name: r.status for r in results} == dict.fromkeys( + assert {r.name: r.status for r in results} == {**dict.fromkeys( ["card", "any_host", "health", "universe_now", "reserved_paths", "data_plane", "clock", "forwarded", - "cookies"], "passed"), results + "cookies"], "passed"), "mcp": "skipped"}, results def test_the_card_declares_the_site_and_its_routes_by_name_and_it_is_ready_through_website_v1( diff --git a/src/agent_env/cli/plugin.py b/src/agent_env/cli/plugin.py index 7f4f2b3f7..11dcd181c 100644 --- a/src/agent_env/cli/plugin.py +++ b/src/agent_env/cli/plugin.py @@ -10,7 +10,7 @@ from contextlib import contextmanager, redirect_stdout from importlib.metadata import PackageNotFoundError, distribution, packages_distributions from pathlib import Path -from typing import Optional +from typing import Optional, TextIO import click @@ -121,6 +121,11 @@ def _quiet() -> Iterator[None]: logger.setLevel(previous) +def _stdout() -> TextIO: + """The stdout a command started with, which its report goes to though a plugin it imports replaces sys.stdout.""" + return click.get_text_stream("stdout") + + def collect(root: click.Group, *, load: bool) -> Report: # A plugin that prints while it is imported must not corrupt `--json` on stdout. with _quiet(), redirect_stdout(sys.stderr): @@ -334,8 +339,9 @@ def plugin(): @click.option("--no-load", is_flag=True, help="Read installed metadata only; import no type plugin.") def list_command(as_json: bool, no_load: bool): """List every installed plugin package, what it provides, and whether it took effect.""" + out = _stdout() report = collect(_root(), load=not no_load) - click.echo(_json(_as_dict(report)) if as_json else render_list(report)) + click.echo(_json(_as_dict(report)) if as_json else render_list(report), file=out) @plugin.command() @@ -344,6 +350,7 @@ def list_command(as_json: bool, no_load: bool): @click.option("--no-load", is_flag=True, help="Read installed metadata only; import no type plugin.") def show(package: str, as_json: bool, no_load: bool): """Show one plugin PACKAGE: each contribution, its status and why.""" + out = _stdout() report = collect(_root(), load=not no_load) matches = [d for d in report.distributions if _normalized(d.name) == _normalized(package)] if not matches: @@ -352,9 +359,9 @@ def show(package: str, as_json: bool, no_load: bool): dist = matches[0] effect = None if no_load else config_effect(dist.name) if as_json: - click.echo(_json({**_as_dict(report), "plugins": [dataclasses.asdict(dist)], "config_effect": effect})) + click.echo(_json({**_as_dict(report), "plugins": [dataclasses.asdict(dist)], "config_effect": effect}), file=out) return - click.echo(render_show(report, dist, effect)) + click.echo(render_show(report, dist, effect), file=out) @plugin.command() @@ -366,6 +373,7 @@ def check(ctx: click.Context, as_json: bool): A replaced contribution is not a failure: config chose it. Meant for CI and image builds. """ + out = _stdout() report = collect(_root(), load=True) problems = [(d, c) for d in report.distributions for c in d.contributions if c.status in _CHECK_FAILS] error = report.inventory.config_error @@ -380,22 +388,22 @@ def check(ctx: click.Context, as_json: bool): **_as_dict(report), "ok": not reasons, "problems": [{"package": d.name, "version": d.version, **dataclasses.asdict(c)} for d, c in problems], - })) + }), file=out) if reasons: ctx.exit(1) return for d, c in problems: - click.echo(f"{d.name} {d.version}: {_KINDS[c.group][0]} {c.name}: {_outcome(c)}") + click.echo(f"{d.name} {d.version}: {_KINDS[c.group][0]} {c.name}: {_outcome(c)}", file=out) if error: - click.echo(_coded(f"config: {error.reason}", error.code)) + click.echo(_coded(f"config: {error.reason}", error.code), file=out) for line in _unreadable(report): - click.echo(line) + click.echo(line, file=out) if reasons: raise click.ClickException("; ".join(reasons)) total = sum(len(d.contributions) for d in report.distributions) replaced = sum(c.status == "replaced" for d in report.distributions for c in d.contributions) note = f"; {replaced} replaced by config" if replaced else ", all in effect" - click.echo(f"ok: {len(report.distributions)} plugin package(s), {total} contribution(s){note}") + click.echo(f"ok: {len(report.distributions)} plugin package(s), {total} contribution(s){note}", file=out) @plugin.command() diff --git a/src/agent_env/computers/artifacts/ubuntu-24.04/start-desktop b/src/agent_env/computers/artifacts/ubuntu-24.04/start-desktop index d3dbf99f9..208dbe570 100755 --- a/src/agent_env/computers/artifacts/ubuntu-24.04/start-desktop +++ b/src/agent_env/computers/artifacts/ubuntu-24.04/start-desktop @@ -15,6 +15,9 @@ case "${1:-}" in exec /usr/bin/supervisord -c /etc/supervisor/supervisord.conf ;; xvfb) + # supervisord runs one X at a time, so a lock here is one a killed X left, and X refuses to start while the pid + # it names is taken, which after a restart it often is. + rm -f "/tmp/.X${DISPLAY#:}-lock" exec Xvfb "$DISPLAY" -screen 0 "${resolution}x24" -nolisten tcp -ac ;; with-x) diff --git a/src/agent_env/env/gateway/entrypoint.py b/src/agent_env/env/gateway/entrypoint.py index 65cf7e3c3..7f3606e17 100644 --- a/src/agent_env/env/gateway/entrypoint.py +++ b/src/agent_env/env/gateway/entrypoint.py @@ -3,12 +3,13 @@ import logging import os import time +from urllib.parse import urlsplit import httpx from . import AGENT_ENV_GATEWAY_MCP_PORT, GatewayMode, InternalMCPServer -from .constants import DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV -from .gateway import Gateway +from .constants import DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV, MCP_TRANSPORT, WELL_KNOWN_PATH +from .gateway import Gateway, card_url logger = logging.getLogger(__name__) @@ -45,6 +46,7 @@ def parse_internal_servers(spec: str, port: int) -> list[InternalMCPServer]: # docker-compose `depends_on`), so the old 30s ceiling was far too short. _INTERNAL_SERVER_TIMEOUT_S = int(os.environ.get("GATEWAY_INTERNAL_SERVER_TIMEOUT_S", "180")) _INTERNAL_SERVER_POLL_S = 2 +_UNREAD = object() def _ready_url(mcp_url: str) -> str: @@ -103,6 +105,64 @@ async def wait_for_internal_servers(servers: list[InternalMCPServer]) -> None: logger.info(f"All {len(servers)} internal MCP server(s) ready in {time.monotonic() - t0:.1f}s") +async def website_mcp_servers(sites: dict[str, str], taken: frozenset[str] = frozenset(), + timeout_s: int = _INTERNAL_SERVER_TIMEOUT_S) -> list[InternalMCPServer]: + """The sites, by name at the address their server answers, that are MCP servers too: each as an internal server at + the MCP endpoint its card declares, so the gateway serves its tools as it does an MCP env's. A site whose card + declares no MCP interface, or that answers no card, is left out, as is one named as an internal server already + (``taken``). Every site is asked at once, and one whose server can't be reached yet is asked again until + ``timeout_s``; raises, naming them, when a site's card still can't be read then, so the gateway doesn't start + without a site, as with an MCP server that never comes up.""" + asked = {name: url.rstrip("/") for name, url in sites.items() if name not in taken} + deadline = time.monotonic() + timeout_s + async with httpx.AsyncClient() as client: + cards = await asyncio.gather(*(_site_card(client, base, deadline) for base in asked.values())) + unread = [name for name, card in zip(asked, cards) if card is _UNREAD] + if unread: + raise RuntimeError(f"Website card(s) not readable after {timeout_s}s: {', '.join(sorted(unread))}") + found: list[InternalMCPServer] = [] + for (name, base), card in zip(asked.items(), cards): + declared = next((i.get("url") for i in (card or {}).get("additionalInterfaces") or [] + if isinstance(i, dict) and i.get("transport") == MCP_TRANSPORT and i.get("url")), None) + if isinstance(declared, str) and (mcp_url := _site_mcp_url(base, declared)) is not None: + logger.info(f"Site {name} is an MCP server at {mcp_url}") + found.append(InternalMCPServer(name=name, mcp_url=mcp_url)) + elif declared is not None: + logger.warning(f"Site {name}'s card declares its MCP endpoint at {declared!r}, not on its own server; its " + "tools aren't served") + return found + + +def _site_mcp_url(base: str, declared: str) -> str | None: + """Where the site at ``base`` serves the MCP endpoint its card declares at ``declared`` (``card_url``), on the + address's scheme and host as the address spells them, which the gateway's /svc routes use; None for another + server.""" + resolved = card_url(base, declared) + if resolved is None or not resolved.own_server: + return None + own = urlsplit(base.rstrip("/")) + return f"{own.scheme}://{own.netloc}{resolved.path}" + (f"?{resolved.query}" if resolved.query else "") + + +async def _site_card(client: httpx.AsyncClient, base: str, deadline: float) -> dict | None | object: + """The card the site at ``base`` serves, None when it serves none; asked again while its server can't be reached + or answers 5xx, until ``deadline``, then ``_UNREAD``.""" + while True: + try: + answer = await client.get(base + WELL_KNOWN_PATH, timeout=5) + except httpx.HTTPError: + answer = None + if answer is not None and answer.status_code < 500: + try: + card = answer.json() if answer.status_code == 200 else None + except ValueError: + return None + return card if isinstance(card, dict) else None + if time.monotonic() >= deadline: + return _UNREAD + await asyncio.sleep(_INTERNAL_SERVER_POLL_S) + + def parse_name_url_map(env_str: str) -> dict[str, str]: # Expected format: 'slack=http://slack-website-backend:8000,email=http://email-website-backend:8000' urls: dict[str, str] = {} @@ -137,10 +197,13 @@ def parse_name_url_map(env_str: str) -> dict[str, str]: spec=os.environ.get("INTERNAL_MCP_SERVERS", ""), port=int(os.environ.get("INTERNAL_MCP_PORT", str(AGENT_ENV_GATEWAY_MCP_PORT))), ) - asyncio.run(wait_for_internal_servers(internal_mcp_servers)) - website_urls = parse_name_url_map(os.environ.get("WEBSITE_URLS", "")) rest_proxy_urls = parse_name_url_map(os.environ.get("REST_PROXY_URLS", "")) + internal_mcp_servers += asyncio.run(website_mcp_servers( + {name: rest_proxy_urls.get(name, url) for name, url in website_urls.items()}, + taken=frozenset(server.name for server in internal_mcp_servers))) + asyncio.run(wait_for_internal_servers(internal_mcp_servers)) + gateway_mode = GatewayMode(os.environ.get("GATEWAY_MODE", GatewayMode.PERFORMANCE.value)) service_db_url = os.environ.get("SERVICE_DB_URL") gateway = Gateway( diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index 79f6d1911..8fec7558d 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -14,8 +14,8 @@ from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path -from typing import TYPE_CHECKING, Any -from urllib.parse import parse_qsl +from typing import TYPE_CHECKING, Any, NamedTuple +from urllib.parse import parse_qsl, urljoin, urlsplit from uuid import uuid4 import anyio @@ -153,6 +153,7 @@ def __init__( self.internal_mcp_servers = internal_mcp_servers self.website_urls = website_urls or {} self.rest_proxy_urls = rest_proxy_urls or {} + self._site_mcp_endpoints = site_mcp_endpoints(internal_mcp_servers, self.website_urls, self.rest_proxy_urls) self._site_reserved: dict[str, frozenset[str]] = {} self._withheld = frozenset(name.strip().lower().encode() for name in provider_headers if name.strip()) self._browsers = BrowserRoutes(self.rest_proxy_urls, self.REST_PROXY_TIMEOUT, self._log_event, @@ -651,12 +652,14 @@ async def _proxy_rest_request(self, request: Request, url_map: dict[str, str] | if base_url is None: return self._json_response({"error": f"Unknown service: {service_name}"}, 404) role = self._request_role(request) - mcp_server = url_map is self.rest_proxy_urls and service_name.startswith("mcp-") + site_mcp = self._site_mcp_endpoints.get(service_name) if url_map is self.rest_proxy_urls else None + mcp_server = url_map is self.rest_proxy_urls and (service_name.startswith("mcp-") or site_mcp is not None) + mcp_endpoint = site_mcp or "mcp" if url_map is self.rest_proxy_urls and (cdp := await self._browsers.cdp_answer(request, base_url, role)) is not None: return cdp if request.method in _SAFE_METHODS: return await forward(request, base_url, role=role, timeout=self.REST_PROXY_TIMEOUT, mcp_server=mcp_server, - withheld=self._withheld) + withheld=self._withheld, mcp_endpoint=mcp_endpoint) changelog_before = self._query_changelog_id() attribution = self._attribution(role) # at the start: a fire may be gone by the time the response ends @@ -681,7 +684,7 @@ async def log_write(status: int) -> None: }) return await forward(request, base_url, role=role, timeout=self.REST_PROXY_TIMEOUT, mcp_server=mcp_server, - on_done=log_write, withheld=self._withheld) + on_done=log_write, withheld=self._withheld, mcp_endpoint=mcp_endpoint) async def _proxy_website_request(self, request: Request) -> Response: """Proxy a website request; with a website or state trigger registered, feed it to the engine and hold barriers. @@ -797,7 +800,8 @@ async def _site_reserved_paths(self, app: str) -> frozenset[str] | None: card = response.json() if response.is_success else None except ValueError: card = None - self._site_reserved[app] = site_reserved_paths(card) if isinstance(card, dict) else frozenset() + address = self.website_urls.get(app) or base_url + self._site_reserved[app] = site_reserved_paths(card, address) if isinstance(card, dict) else frozenset() return self._site_reserved[app] def _backing_keys(self) -> list[str]: @@ -1668,17 +1672,80 @@ def reserved_site_path(path: str, reserved: frozenset[str] = frozenset()) -> boo for base in (RPC_PATH, *reserved)) -def site_reserved_paths(card: dict) -> frozenset[str]: - """The paths a site's card advertises for agentenv, which its route never reaches: its ``url``, and every - extension's endpoint and each of its methods'; each one a path of its own, not the site's root.""" - paths = [card.get("url")] +def site_mcp_endpoints(servers: list[InternalMCPServer], website_urls: dict[str, str], + rest_proxy_urls: dict[str, str]) -> dict[str, str]: + """The MCP endpoint, as a path under its /svc// base, of each site that is an internal MCP server too, by + name: where a call through /svc// gets the gateway's role in its JSON-RPC ``_meta``, as an MCP server's does.""" + endpoints = {} + for server in servers: + base = rest_proxy_urls.get(server.name, "") + if server.name not in website_urls or not base: + continue + resolved = card_url(base, server.mcp_url) + if resolved is not None and resolved.own_server and resolved.route_path is not None: + endpoints[server.name] = resolved.route_path + return endpoints + + +class CardURL(NamedTuple): + """A URL a site's card names, as the site at an address serves it: its path and query; the path a request on the + site's route reaches it at, its path under the address's (None when it isn't under it); and whether it is on the + site's own server.""" + path: str + query: str + route_path: str | None + own_server: bool + + +def card_url(base: str, url: str) -> CardURL | None: + """``url``, which a site's card names, against the site's address ``base`` (a URL, or only its path): a path, + absolute or relative, joined under the address's path, since a card's paths are relative to where it was read, and + on the site's own server; a full URL as it is, on the site's own server when its scheme, host and port are the + address's. The one reading of a card's URLs that discovery, /svc role stamping and a route's reserved paths share. + None for one with no path, or that can't be read.""" + try: + own, given = urlsplit(base.rstrip("/")), urlsplit(url) + full = bool(given.scheme or given.netloc) + same = not full or _address(given) == _address(own) + except ValueError: + return None + under = own.path.rstrip("/") + if full: + path = given.path + elif given.path.startswith("/"): + path = under + given.path + else: + path = urljoin(under + "/", given.path) if given.path else "" + if not path.startswith("/"): + return None + normalized = _normalized(path) + route = (normalized[len(under):] or "/") if normalized == under or normalized.startswith(under + "/") else None + return CardURL(path, given.query, route, same) + + +def _address(url) -> tuple: + """A split URL's scheme, host and port, its scheme's default port when it names none.""" + return url.scheme, url.hostname, url.port or {"http": 80, "https": 443}.get(url.scheme) + + +def site_reserved_paths(card: dict, base: str = "") -> frozenset[str]: + """The paths a site's card advertises for agentenv, which its route never reaches: its ``url``, every interface's, + such as the MCP endpoint of a site that is an MCP server too, and every extension's endpoint and each of its + methods'; each one a path of its own, not the site's root, read from a full URL or one with a query as well as from + a bare path, as ``card_url`` reads them against the site's address ``base``. A full URL's path is reserved + whatever its host, since a site is reached under names it never sees in advance: as its route reaches it, under the + address's path, or as it is when it isn't under it.""" + urls = [card.get("url")] + urls += [i.get("url") for i in card.get("additionalInterfaces") or [] if isinstance(i, dict)] for ext in (card.get("capabilities") or {}).get("extensions") or []: params = ext.get("params") if isinstance(ext, dict) else None if not isinstance(params, dict): continue - paths += [endpoint for *_, endpoint in operations(params)] - paths.append(params.get("endpoint")) - return frozenset(_normalized(p) for p in paths if isinstance(p, str) and p.startswith("/") and _normalized(p) != "/") + urls += [endpoint for *_, endpoint in operations(params)] + urls.append(params.get("endpoint")) + resolved = (card_url(base, url) for url in urls if isinstance(url, str)) + paths = (_normalized(r.path if r.route_path is None else r.route_path) for r in resolved if r is not None) + return frozenset(p for p in paths if p != "/") def site_request_path(request: HTTPConnection) -> str | None: diff --git a/src/agent_env/env/gateway/rest_proxy.py b/src/agent_env/env/gateway/rest_proxy.py index 6a2e00edc..54235b290 100644 --- a/src/agent_env/env/gateway/rest_proxy.py +++ b/src/agent_env/env/gateway/rest_proxy.py @@ -86,11 +86,11 @@ async def forward(request: Request, base_url: str, *, role: str, timeout: httpx.Timeout, mcp_server: bool, on_done: Callable[[int], Awaitable[None]] | None = None, - withheld: frozenset[bytes] = frozenset()) -> Response: + withheld: frozenset[bytes] = frozenset(), mcp_endpoint: str = "mcp") -> Response: """Forward `request` to the service at `base_url` as `role`, without the headers named in `withheld`, and stream the response through. `mcp_server` says - the service is a backing MCP server, whose own endpoint takes the role in its JSON-RPC `_meta` as well as in the - header. A 400 names a request-target the proxy cannot route; a 502 names an upstream that never answered. + the service is a backing MCP server, whose own endpoint, `mcp_endpoint` under it, takes the role in its JSON-RPC + `_meta` as well as in the header. A 400 names a request-target the proxy cannot route; a 502 names an upstream that never answered. `on_done` is awaited with the status once the response has ended, after its last byte when streamed.""" try: target_url = upstream_url(base_url, request) @@ -98,7 +98,7 @@ async def forward(request: Request, base_url: str, *, role: str, timeout: httpx. return await _done(_json_response({"error": str(e)}, 400), on_done) content = await request.body() # /svc and /website request bodies are small (JSON-RPC, forms) headers = upstream_headers(request, role, withheld=withheld) - if mcp_server and request.method == "POST" and is_child_mcp_path(request.path_params.get("path", "")): + if mcp_server and request.method == "POST" and is_child_mcp_path(request.path_params.get("path", ""), mcp_endpoint): # The gateway is the trust boundary for identity: a child's proxied MCP endpoint stays reachable, but the # role in its JSON-RPC `_meta` is the gateway's to write, like the header above. content = stamp_jsonrpc_role(content, role) @@ -200,15 +200,19 @@ def route_prefix(request: HTTPConnection) -> str: return raw_path.decode("latin-1").rstrip("/") -def is_child_mcp_path(decoded_rest: str) -> bool: - """Whether a /svc/{service}/{path} remainder reaches the backing server's MCP endpoint (`mcp` or anything under - it), judged after dot-segment normalisation, since httpx collapses `x/../mcp` to `/mcp` on the way upstream. +def is_child_mcp_path(decoded_rest: str, endpoint: str = "mcp") -> bool: + """Whether a /svc/{service}/{path} remainder reaches the backing server's MCP endpoint (`endpoint`, `mcp` unless + its card declares another, or anything under it), judged after dot-segment normalisation, since httpx collapses + `x/../mcp` to `/mcp` on the way upstream. >>> [is_child_mcp_path(p) for p in ("mcp", "mcp/", "mcp/x", "x/../mcp", "./mcp", "/mcp", "mcpx", "agentenv", "")] [True, True, True, True, True, True, False, False, False] + >>> [is_child_mcp_path(p, "/tools/mcp/") for p in ("tools/mcp", "tools/x/../mcp", "mcp")] + [True, True, False] """ normalized = posixpath.normpath("/" + decoded_rest.lstrip("/")) - return normalized == "/mcp" or normalized.startswith("/mcp/") + at = posixpath.normpath("/" + endpoint.strip("/")) + return normalized == at or normalized.startswith(at + "/") def stamp_jsonrpc_role(body: bytes, role: str) -> bytes: diff --git a/tst/integration/computer/base_computer_test.py b/tst/integration/computer/base_computer_test.py index 3920ea10e..4d3d3f83b 100644 --- a/tst/integration/computer/base_computer_test.py +++ b/tst/integration/computer/base_computer_test.py @@ -84,6 +84,20 @@ def test_the_desktop_comes_back_however_often_x_dies(base): assert "window id #" in wm +def test_the_desktop_comes_back_from_a_restart_that_killed_x_with_its_lock_naming_a_running_pid(base): + """A restart gives pids out again, so the lock a killed X left can name a running process; here tini, pid 1.""" + name = f"agentenv-test-base-{uuid.uuid4().hex[:8]}" + docker("run", "-d", "--name", name, base) + try: + _all_running(name) + _exec(name, "sh", "-c", 'printf "%10d\\n" 1 > /tmp/lock && chmod 444 /tmp/lock && mv -f /tmp/lock /tmp/.X99-lock') + docker("restart", "-t", "0", name) + running = _all_running(name) + finally: + subprocess.run(["docker", "rm", "-f", name], capture_output=True) + assert all(" RUNNING " in line for line in running.splitlines()), running + + @pytest.mark.parametrize("setting, value, error", [ ("RESOLUTION", "big", "resolution is x, each 1 to 32767 pixels, not 'big'"), ("RESOLUTION", "40000x800", "resolution is x, each 1 to 32767 pixels, not '40000x800'"), diff --git a/tst/integration/computer/recording_chaos_test.py b/tst/integration/computer/recording_chaos_test.py index a2ffb8583..6fe83a82e 100644 --- a/tst/integration/computer/recording_chaos_test.py +++ b/tst/integration/computer/recording_chaos_test.py @@ -128,17 +128,21 @@ async def _ready(name: str) -> Driver: """A client for the container's driver once its computer is ready, within the time a deploy gives one; a restart may move its port.""" deadline = time.monotonic() + READY_TIMEOUT_S + status = None while time.monotonic() < deadline: port = subprocess.run(["docker", "port", name, str(PORT)], capture_output=True, text=True).stdout if port: try: driver = await Driver.connect(f"http://{port.splitlines()[0]}", 30) - if (await driver.invoke(COMPUTER_READY, "status"))["ready"]: + status = await driver.invoke(COMPUTER_READY, "status") + if status["ready"]: return driver except Exception: # not serving yet pass await asyncio.sleep(0.25) - raise TimeoutError(f"{name}'s computer wasn't ready within {READY_TIMEOUT_S:.0f} s") + failing = [check for check in (status or {}).get("checks", []) if not check["ok"]] + raise TimeoutError(f"{name}'s computer wasn't ready within {READY_TIMEOUT_S:.0f} s: " + f"{failing or 'its driver never answered'}") async def _layout(driver: Driver) -> dict: diff --git a/tst/integration/env/envs/server_env_local_test.py b/tst/integration/env/envs/server_env_local_test.py index cf63b3673..1285ea568 100644 --- a/tst/integration/env/envs/server_env_local_test.py +++ b/tst/integration/env/envs/server_env_local_test.py @@ -39,6 +39,7 @@ from agent_env.task_step.context import TaskStepContext from agent_env.task.teardown import TORN_DOWN_KEY from agent_env.task_step.task_steps.teardown_sandboxes import TeardownSandboxesTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.util.a2a_test_agent import put_test_agent pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] @@ -75,7 +76,7 @@ def local_stack(monkeypatch, tmp_path): pytest.skip("docker daemon not available") reg_port = _free_port() reg_name = f"server-e2e-registry-{uuid.uuid4().hex[:8]}" - if _docker("run", "-d", "--rm", "--name", reg_name, "-p", f"127.0.0.1:{reg_port}:5000", "registry:2").returncode != 0: + if _docker("run", "-d", "--rm", "--name", reg_name, "-p", f"127.0.0.1:{reg_port}:5000", REGISTRY_IMAGE).returncode != 0: pytest.skip("could not start registry:2") host = f"localhost:{reg_port}" if not _wait_ready(f"http://{host}/v2/", (200, 401)): diff --git a/tst/integration/store/local_image_composition_test.py b/tst/integration/store/local_image_composition_test.py index 66ddb522e..5404f1d42 100644 --- a/tst/integration/store/local_image_composition_test.py +++ b/tst/integration/store/local_image_composition_test.py @@ -23,6 +23,7 @@ from agent_env.config import configure, reset_config, set_image_store from agent_env.config.paths import state_root from agent_env.store.image_store import LocalRegistryImageStore +from tst.integration.utils.registry import REGISTRY_IMAGE pytestmark = pytest.mark.integration @@ -43,7 +44,7 @@ def _docker(*args): def all_local(monkeypatch, tmp_path): port = _free_port() name = f"local-composition-registry-{uuid.uuid4().hex[:8]}" - start = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", "registry:2") + start = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", REGISTRY_IMAGE) if start.returncode != 0: pytest.skip(f"could not start registry:2: {start.stderr.strip()}") host = f"localhost:{port}" diff --git a/tst/integration/store/local_parallel_deploy_test.py b/tst/integration/store/local_parallel_deploy_test.py index 4964e80c4..322bf7ea1 100644 --- a/tst/integration/store/local_parallel_deploy_test.py +++ b/tst/integration/store/local_parallel_deploy_test.py @@ -27,6 +27,7 @@ from agent_env.task import Task from agent_env.task.store import get_task_instance_store from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.task import journal_invariants as journal pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] @@ -71,9 +72,9 @@ def local_stack_at(monkeypatch: pytest.MonkeyPatch, path: Path): reg_port = _free_port() reg_name = f"par-e2e-registry-{uuid.uuid4().hex[:8]}" - if _docker("run", "-d", "--rm", "--name", reg_name, - "-p", f"127.0.0.1:{reg_port}:5000", "registry:2").returncode != 0: - pytest.skip("could not start registry:2") + start = _docker("run", "-d", "--rm", "--name", reg_name, "-p", f"127.0.0.1:{reg_port}:5000", REGISTRY_IMAGE) + if start.returncode != 0: + pytest.skip(f"could not start registry:2: {start.stderr.strip()}") host = f"localhost:{reg_port}" if not _wait_ready(f"http://{host}/v2/", (200, 401)): _docker("rm", "-f", reg_name) diff --git a/tst/integration/store/local_registry_image_store_test.py b/tst/integration/store/local_registry_image_store_test.py index 0d765a03c..16b94261e 100644 --- a/tst/integration/store/local_registry_image_store_test.py +++ b/tst/integration/store/local_registry_image_store_test.py @@ -13,6 +13,7 @@ import pytest from agent_env.store import LocalRegistryImageStore +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.store import image_conformance pytestmark = pytest.mark.integration @@ -31,7 +32,7 @@ def registry_host(): port = _free_port() name = f"imagestore-conf-registry-{uuid.uuid4().hex[:8]}" start = subprocess.run( - ["docker", "run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", "registry:2"], + ["docker", "run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", REGISTRY_IMAGE], capture_output=True, text=True, ) if start.returncode != 0: diff --git a/tst/integration/store/local_task_lifecycle_test.py b/tst/integration/store/local_task_lifecycle_test.py index fa299e601..09a40d3e7 100644 --- a/tst/integration/store/local_task_lifecycle_test.py +++ b/tst/integration/store/local_task_lifecycle_test.py @@ -41,6 +41,7 @@ from agent_env.store.image_store import LocalRegistryImageStore from agent_env.task import Task from agent_env.task_step.task_steps.deploy_sandbox import DeploySandboxTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE pytestmark = pytest.mark.integration @@ -120,7 +121,7 @@ def _wait_ready(url: str, codes: tuple[int, ...], timeout: int = 45) -> bool: def all_local(monkeypatch, tmp_path): reg_port = _free_port() reg_name = f"local-e2e-registry-{uuid.uuid4().hex[:8]}" - start = _docker("run", "-d", "--rm", "--name", reg_name, "-p", f"127.0.0.1:{reg_port}:5000", "registry:2") + start = _docker("run", "-d", "--rm", "--name", reg_name, "-p", f"127.0.0.1:{reg_port}:5000", REGISTRY_IMAGE) if start.returncode != 0: pytest.skip(f"could not start registry:2: {start.stderr.strip()}") host = f"localhost:{reg_port}" diff --git a/tst/integration/store/oci_registry_image_store_test.py b/tst/integration/store/oci_registry_image_store_test.py index 1a8ce4c0a..4e81b1f89 100644 --- a/tst/integration/store/oci_registry_image_store_test.py +++ b/tst/integration/store/oci_registry_image_store_test.py @@ -19,6 +19,7 @@ OciRegistryImageStore, SecretStoreCredentials, ) +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.store import image_conformance pytestmark = pytest.mark.integration @@ -70,7 +71,7 @@ def authenticated_registry_host(tmp_path_factory): "REGISTRY_AUTH_HTPASSWD_REALM=Registry Realm", "-e", "REGISTRY_AUTH_HTPASSWD_PATH=/auth/htpasswd", - "registry:2", + REGISTRY_IMAGE, ) if start.returncode != 0: pytest.fail(f"could not start authenticated registry:2: {start.stderr.strip()}") diff --git a/tst/integration/task_step/test_container_sandbox_files_local.py b/tst/integration/task_step/test_container_sandbox_files_local.py index 68b84054d..00cfd9fa6 100644 --- a/tst/integration/task_step/test_container_sandbox_files_local.py +++ b/tst/integration/task_step/test_container_sandbox_files_local.py @@ -28,6 +28,7 @@ from agent_env.task_step.task_steps.deploy_sandbox import DeploySandboxTaskStep from agent_env.task_step.task_steps.load_artifact import LoadArtifactTaskStep from agent_env.task_step.task_steps.verifiers.verify_sandbox import VerifySandboxTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.util.a2a_test_agent import put_test_agent pytestmark = pytest.mark.integration @@ -53,7 +54,7 @@ def local_registry(monkeypatch, tmp_path): monkeypatch.setenv("AGENT_ENV_LOCAL_SANDBOX_DIR", str(sandboxes)) port = _free_port() name = f"files-local-registry-{uuid.uuid4().hex[:8]}" - started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", "registry:2") + started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", REGISTRY_IMAGE) assert started.returncode == 0, started.stderr host = f"localhost:{port}" deadline = time.time() + 45 diff --git a/tst/integration/task_step/test_prompt_agent_file_parts_local.py b/tst/integration/task_step/test_prompt_agent_file_parts_local.py index c8bdc57f0..bb2ff6e66 100644 --- a/tst/integration/task_step/test_prompt_agent_file_parts_local.py +++ b/tst/integration/task_step/test_prompt_agent_file_parts_local.py @@ -27,6 +27,7 @@ from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.deploy_human_agent import DeployHumanAgentTaskStep from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.util.a2a_test_agent import AGENT_DIR, put_test_agent pytestmark = pytest.mark.integration @@ -56,7 +57,7 @@ def local_registry(monkeypatch, tmp_path): monkeypatch.setenv("AGENT_ENV_LOCAL_SANDBOX_DIR", str(sandboxes)) port = _free_port() name = f"file-parts-registry-{uuid.uuid4().hex[:8]}" - started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", "registry:2") + started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", REGISTRY_IMAGE) assert started.returncode == 0, started.stderr host = f"localhost:{port}" deadline = time.time() + REGISTRY_READY_SECONDS diff --git a/tst/integration/task_step/test_prompt_agent_live_trajectory_local.py b/tst/integration/task_step/test_prompt_agent_live_trajectory_local.py index b82bd60f0..8c1c1ccbf 100644 --- a/tst/integration/task_step/test_prompt_agent_live_trajectory_local.py +++ b/tst/integration/task_step/test_prompt_agent_live_trajectory_local.py @@ -22,6 +22,7 @@ from agent_env.task import Task from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE from tst.util.a2a_test_agent import put_test_agent pytestmark = pytest.mark.integration @@ -52,7 +53,7 @@ def local_registry(monkeypatch, tmp_path): monkeypatch.setenv("AGENT_ENV_LOCAL_SANDBOX_DIR", str(sandboxes)) port = _free_port() name = f"live-trajectory-registry-{uuid.uuid4().hex[:8]}" - started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", "registry:2") + started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", REGISTRY_IMAGE) assert started.returncode == 0, started.stderr host = f"localhost:{port}" deadline = time.time() + REGISTRY_READY_SECONDS diff --git a/tst/integration/task_step/test_run_code_named_sandbox_local.py b/tst/integration/task_step/test_run_code_named_sandbox_local.py index 7c8b98f89..ab6f13793 100644 --- a/tst/integration/task_step/test_run_code_named_sandbox_local.py +++ b/tst/integration/task_step/test_run_code_named_sandbox_local.py @@ -31,6 +31,7 @@ from agent_env.task_step.task_steps.deploy_sandbox import DeploySandboxTaskStep from agent_env.task_step.task_steps.load_artifact import LoadArtifactTaskStep from agent_env.task_step.task_steps.run_code import RunCodeTaskStep +from tst.integration.utils.registry import REGISTRY_IMAGE pytestmark = pytest.mark.integration @@ -151,7 +152,7 @@ def _free_port() -> int: def local_registry(local_backends): port = _free_port() name = f"run-code-registry-{uuid.uuid4().hex[:8]}" - started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", "registry:2") + started = _docker("run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:5000", REGISTRY_IMAGE) assert started.returncode == 0, started.stderr host = f"localhost:{port}" deadline = time.time() + 45 diff --git a/tst/integration/utils/registry.py b/tst/integration/utils/registry.py new file mode 100644 index 000000000..6ddf19de0 --- /dev/null +++ b/tst/integration/utils/registry.py @@ -0,0 +1,4 @@ +"""The registry image the integration tests' throwaway registries run: the mirror the local-backends workflow pulls before +the tests, so a test starts one without pulling from Docker Hub.""" + +REGISTRY_IMAGE = "mirror.gcr.io/library/registry:2" diff --git a/tst/unit/cli/plugin_report_test.py b/tst/unit/cli/plugin_report_test.py index 5d5b74dff..231886ace 100644 --- a/tst/unit/cli/plugin_report_test.py +++ b/tst/unit/cli/plugin_report_test.py @@ -10,6 +10,8 @@ """ import ast +import importlib +import io import json import os import re @@ -32,6 +34,7 @@ from tst.unit.plugins_test import _EP, _HERE, _Dist, _fresh, _install, _use_config # noqa: F401 (_fresh: autouse) _THIS = "tst.unit.cli.plugin_report_test" +_PLUGIN_CLI = importlib.import_module("agent_env.cli.plugin") _GOLDEN = Path(__file__).parent / "golden" / "plugin_report" _REGENERATE = os.environ.get("AGENT_ENV_REGENERATE_GOLDENS") == "1" _COMMANDS = cli_plugins_mod.CLI_PLUGINS_GROUP @@ -205,6 +208,18 @@ def _report(case: str, monkeypatch, tmp_path) -> dict: return json.loads(result.stdout) + +@pytest.mark.parametrize("case", ["list", "show", "check"]) +def test_the_report_reaches_the_commands_stdout_though_a_plugin_replaces_sys_stdout(case, monkeypatch, tmp_path): + real = _PLUGIN_CLI.environment + + def replacing(*args): + sys.stdout = io.StringIO() + return real(*args) + + monkeypatch.setattr(_PLUGIN_CLI, "environment", replacing) + assert _report(case, monkeypatch, tmp_path)["format_version"] == FORMAT_VERSION + def _normalized(node, tmp_path: Path): """``node`` with machine-specific values and free text replaced by placeholders.""" if isinstance(node, dict): diff --git a/tst/unit/env/gateway/website_mcp_test.py b/tst/unit/env/gateway/website_mcp_test.py new file mode 100644 index 000000000..8f24864ed --- /dev/null +++ b/tst/unit/env/gateway/website_mcp_test.py @@ -0,0 +1,200 @@ +"""A website env that is an MCP server too: the gateway finds its MCP endpoint on its card and serves its tools as it +does an MCP env's, while the site's route keeps that endpoint, like the rest of what its card advertises, out of an +agent's browsing reach.""" +from __future__ import annotations + +import httpx +import pytest +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client +from pytest_socket import enable_socket + +from agent_env.env.gateway import AGENT_ENV_ROLE_HEADER, AGENT_ENV_ROLE_META_KEY, InternalMCPServer +from agent_env.env.gateway import entrypoint as entrypoint_module +from agent_env.env.gateway.entrypoint import website_mcp_servers +from agent_env.env.gateway.gateway import Gateway, site_mcp_endpoints, site_reserved_paths +from agentenv_protocol import DataPart, ToolContext, environment_card, get_data, tool +from agentenv_protocol.website import WebsiteEnvironment, website +from tst.unit.env.gateway.process_scope_socket_test import _serve + +CARD = "/.well-known/agent-env.json" + + +@pytest.mark.parametrize("declared", ["/mcp", "http://shop:8000/mcp", "/mcp?transport=http", "http://shop:8000/mcp/#x", + "mcp", "./mcp", "mcp?transport=http"]) +def test_a_sites_mcp_interface_is_one_of_its_reserved_paths_however_its_card_writes_it(declared): + card = {"url": "/agentenv", "additionalInterfaces": [{"url": declared, "transport": "mcp"}, {"transport": "x"}]} + assert site_reserved_paths(card) == frozenset({"/agentenv", "/mcp"}) + + +def test_a_relative_mcp_interface_is_reserved_under_the_sites_root_as_discovery_resolves_it(): + card = {"additionalInterfaces": [{"url": "tools/mcp", "transport": "mcp"}]} + assert site_reserved_paths(card) == site_reserved_paths(card, "/base") == frozenset({"/tools/mcp"}) + + +def test_a_sites_mcp_endpoint_is_its_internal_servers_path_under_its_svc_base(): + servers = [InternalMCPServer("shop", "http://shop-backend:8000/tools/mcp?transport=http"), + InternalMCPServer("slack", "http://slack:18765/mcp"), + InternalMCPServer("odd", "http://elsewhere:9/mcp")] + sites = {"shop": "http://shop-backend:8000", "odd": "http://odd-backend:8000"} + assert site_mcp_endpoints(servers, sites, {**sites, "mcp-slack": "http://slack:18765"}) == {"shop": "/tools/mcp"} + spelled = [InternalMCPServer("tls", "https://TLS.test:443/mcp"), InternalMCPServer("deep", "http://d:80/base/mcp")] + bases = {"tls": "https://tls.test", "deep": "http://d/base/"} + assert site_mcp_endpoints(spelled, bases, bases) == {"tls": "/mcp", "deep": "/mcp"} + + +@pytest.mark.parametrize(("base", "declared", "mcp_url"), [ + ("https://shop.test", "https://shop.test:443/mcp", "https://shop.test/mcp"), + ("https://shop.test", "https://SHOP.test/mcp?transport=http", "https://shop.test/mcp?transport=http"), + ("http://shop:8000", "/mcp", "http://shop:8000/mcp"), + ("http://shop:8000", "http://shop:8001/mcp", None), + ("http://shop:8000", "https://shop:8000/mcp", None), + ("http://shop/base", "http://shop/base/mcp", "http://shop/base/mcp"), + ("http://shop/base", "/mcp", "http://shop/base/mcp"), +]) +def test_a_declared_mcp_url_on_the_sites_server_keeps_the_sites_own_spelling(base, declared, mcp_url): + assert entrypoint_module._site_mcp_url(base, declared) == mcp_url + if mcp_url is not None: + servers = [InternalMCPServer("shop", mcp_url)] + assert site_mcp_endpoints(servers, {"shop": base}, {"shop": base}) == {"shop": "/mcp"} + + +@pytest.mark.asyncio +async def test_a_site_behind_a_path_prefix_is_found_at_the_full_url_its_card_declares(monkeypatch): + def site(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/base" + CARD + return httpx.Response(200, json={"additionalInterfaces": [{"url": "http://shop/base/mcp", "transport": "mcp"}]}) + + real = httpx.AsyncClient + monkeypatch.setattr(entrypoint_module.httpx, "AsyncClient", lambda **kw: real(transport=httpx.MockTransport(site), **kw)) + assert await website_mcp_servers({"shop": "http://shop/base"}) == [InternalMCPServer("shop", "http://shop/base/mcp")] + + +@pytest.mark.asyncio +async def test_the_sites_whose_cards_declare_an_mcp_interface_are_internal_servers(monkeypatch): + calls: dict[str, int] = {} + cards = { + "shop": {"additionalInterfaces": [{"url": "/tools/mcp", "transport": "mcp"}]}, + "mail": {"additionalInterfaces": []}, + "other": {"additionalInterfaces": [{"url": "/rpc", "transport": "grpc"}]}, + "full": {"additionalInterfaces": [{"url": "http://full:8000/tools/mcp", "transport": "mcp"}]}, + "away": {"additionalInterfaces": [{"url": "http://elsewhere:8000/mcp", "transport": "mcp"}]}, + } + + def site(request: httpx.Request) -> httpx.Response: + host = request.url.host + calls[host] = calls.get(host, 0) + 1 + assert request.url.path == CARD + if host == "late" and calls[host] == 1: + raise httpx.ConnectError("not up yet") + if host == "late": + return httpx.Response(200, json={"additionalInterfaces": [{"url": "/mcp", "transport": "mcp"}]}) + if host == "nocard": + return httpx.Response(404) + if host == "html": + return httpx.Response(200, text="") + return httpx.Response(200, json=cards[host]) + + real = httpx.AsyncClient + monkeypatch.setattr(entrypoint_module.httpx, "AsyncClient", lambda **kw: real(transport=httpx.MockTransport(site), **kw)) + monkeypatch.setattr(entrypoint_module, "_INTERNAL_SERVER_POLL_S", 0) + sites = {name: f"http://{name}:8000/" for name in ("shop", "mail", "other", "late", "nocard", "html", "full", "away", + "slack")} + + found = await website_mcp_servers(sites, taken=frozenset({"slack"})) + + assert found == [InternalMCPServer("shop", "http://shop:8000/tools/mcp"), InternalMCPServer("late", "http://late:8000/mcp"), + InternalMCPServer("full", "http://full:8000/tools/mcp")] + assert "slack" not in calls and calls["late"] == 2 + + +@pytest.mark.asyncio +async def test_a_site_whose_card_cant_be_read_in_time_stops_the_gateway_starting(monkeypatch): + def down(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("down") + + real = httpx.AsyncClient + monkeypatch.setattr(entrypoint_module.httpx, "AsyncClient", lambda **kw: real(transport=httpx.MockTransport(down), **kw)) + monkeypatch.setattr(entrypoint_module, "_INTERNAL_SERVER_POLL_S", 0) + with pytest.raises(RuntimeError, match="not readable after 0s: shop"): + await website_mcp_servers({"shop": "http://shop:8000"}, timeout_s=0) + + +@pytest.mark.asyncio +async def test_sites_are_asked_at_once_and_the_one_never_up_is_named(monkeypatch): + calls: dict[str, int] = {} + + def site(request: httpx.Request) -> httpx.Response: + host = request.url.host + calls[host] = calls.get(host, 0) + 1 + if host == "down" or calls[host] < 3: + raise httpx.ConnectError("not up yet") + return httpx.Response(200, json={"additionalInterfaces": [{"url": "/mcp", "transport": "mcp"}]}) + + real = httpx.AsyncClient + monkeypatch.setattr(entrypoint_module.httpx, "AsyncClient", lambda **kw: real(transport=httpx.MockTransport(site), **kw)) + monkeypatch.setattr(entrypoint_module, "_INTERNAL_SERVER_POLL_S", 0.01) + with pytest.raises(RuntimeError, match=r"not readable after 1s: down$"): + await website_mcp_servers({"down": "http://down", "late": "http://late"}, timeout_s=1) + assert calls["late"] == 3 + + +@environment_card(name="shop") +@website(domains=["shop.test"]) +class _Shop(WebsiteEnvironment): + def __init__(self) -> None: + super().__init__() + self.orders: list[dict] = [] + + @tool() + def place_order(self, item: str, ctx: ToolContext) -> str: + """Place an order for item, as the caller.""" + self.orders.append({"item": item, "by": ctx.role}) + return f"order {len(self.orders)} placed" + + @get_data + def export(self) -> list: + return [DataPart(data={"orders": self.orders})] + + +@pytest.mark.asyncio +async def test_the_gateway_serves_a_sites_tools_and_its_route_refuses_the_sites_mcp_endpoint(): + enable_socket() + site = _Shop() + site_server, site_url = await _serve(site.create_app()) + try: + servers = await website_mcp_servers({"shop": site_url}) + assert servers == [InternalMCPServer("shop", f"{site_url}/mcp")] + gw = Gateway(host="127.0.0.1", port=0, server_name="t", internal_mcp_servers=servers, + website_urls={"shop": site_url}, rest_proxy_urls={"shop": site_url}) + gw_server, gw_url = await _serve(gw._asgi_app()) + try: + async with ( + httpx.AsyncClient(headers={AGENT_ENV_ROLE_HEADER: "alice@example.com"}, timeout=30) as http, + streamable_http_client(f"{gw_url}/mcp", http_client=http) as (read, write, _), + ClientSession(read, write) as session, + ): + await session.initialize() + assert "place_order" in [t.name for t in (await session.list_tools()).tools] + result = await session.call_tool("place_order", {"item": "milk"}) + assert not result.isError and result.content[0].text == "order 1 placed" + forged = {AGENT_ENV_ROLE_META_KEY: "mallory@example.com"} + async with ( + httpx.AsyncClient(headers={AGENT_ENV_ROLE_HEADER: "bob@example.com"}, timeout=30) as http, + streamable_http_client(f"{gw_url}/svc/shop/mcp", http_client=http) as (read, write, _), + ClientSession(read, write) as session, + ): + await session.initialize() + assert not (await session.call_tool("place_order", {"item": "eggs"}, meta=forged)).isError + async with httpx.AsyncClient(timeout=30) as http: + browsing = await http.post(f"{gw_url}/website/shop/mcp", json={"jsonrpc": "2.0", "id": 1, + "method": "tools/list"}) + data = await http.post(f"{gw_url}/svc/shop/agentenv", json={"jsonrpc": "2.0", "id": 1, + "method": "data/get"}) + finally: + await gw_server.shutdown() + finally: + await site_server.shutdown() + assert site.orders == [{"item": "milk", "by": "alice@example.com"}, {"item": "eggs", "by": "bob@example.com"}] + assert browsing.status_code == 404 + assert data.json()["result"]["parts"][0]["data"]["orders"][0]["item"] == "milk" diff --git a/tst/unit/env/gateway/website_reserved_paths_test.py b/tst/unit/env/gateway/website_reserved_paths_test.py index 306a31d76..40c238684 100644 --- a/tst/unit/env/gateway/website_reserved_paths_test.py +++ b/tst/unit/env/gateway/website_reserved_paths_test.py @@ -62,7 +62,9 @@ async def _streamed(): def test_a_sites_reserved_paths_are_the_ones_its_card_advertises_each_a_path_of_its_own(): assert site_reserved_paths(_CARD) == {"/agentenv", "/agentenv/ext/website/ready", "/grade", "/grading/now", "/hooks"} - assert site_reserved_paths({"name": "x", "url": "http://elsewhere/agentenv"}) == frozenset() + # a full URL's path is reserved whatever its host: the site is reached under names it never sees in advance + assert site_reserved_paths({"name": "x", "url": "http://elsewhere/agentenv?v=1"}) == frozenset({"/agentenv"}) + assert site_reserved_paths({"name": "x", "url": "http://elsewhere"}) == frozenset() reserved = site_reserved_paths(_CARD) for path in ("/grade", "/grade/", "/x/../grade", "/grading/now", "/hooks/a"): assert reserved_site_path(path, reserved), path @@ -70,6 +72,55 @@ def test_a_sites_reserved_paths_are_the_ones_its_card_advertises_each_a_path_of_ assert not reserved_site_path(path, reserved), path + +def test_a_full_url_under_the_sites_address_is_reserved_where_its_route_reaches_it(): + card = {"name": "shop", "additionalInterfaces": [{"url": "http://shop/base/mcp", "transport": "mcp"}]} + assert site_reserved_paths(card, "/base") == frozenset({"/mcp"}) + assert site_reserved_paths(card, "/base/") == frozenset({"/mcp"}) + assert site_reserved_paths(card) == frozenset({"/base/mcp"}) + assert site_reserved_paths({"name": "shop", "url": "/base/agentenv"}, "/base") == frozenset({"/base/agentenv"}) + + +@pytest.mark.asyncio +async def test_a_site_behind_a_base_path_refuses_the_mcp_endpoint_its_card_names_by_full_url(monkeypatch): + reached = [] + card = {"name": "shop", "url": "/agentenv", + "additionalInterfaces": [{"url": "http://shop-website-backend:8000/base/mcp", "transport": "mcp"}]} + + def site(request): + reached.append(request.url.path) + if request.url.path == f"/base{WELL_KNOWN_PATH}": + return httpx.Response(200, json=card) + return httpx.Response(200, content=_streamed()) + + transport = httpx.MockTransport(site) + monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw)) + gw = _gateway(_Changelog(), website_urls={"shop": "http://shop-website-backend:8000/base"}, + rest_proxy_urls={"shop": "http://shop-website-backend:8000/base"}) + + for path in ("mcp", "mcp/", "x/../mcp"): + assert (await _http(gw, "POST", f"/website/shop/{path}", body=b"{}"))[0] == 404, path + assert (await _http(gw, "GET", "/website/shop/orders"))[0] == 200 + assert reached == [f"/base{WELL_KNOWN_PATH}", "/base/orders"] + + +@pytest.mark.asyncio +async def test_a_site_refuses_the_mcp_endpoint_its_card_names_by_a_relative_path(monkeypatch): + card = {"name": "shop", "url": "/agentenv", "additionalInterfaces": [{"url": "mcp", "transport": "mcp"}]} + + def site(request): + if request.url.path == WELL_KNOWN_PATH: + return httpx.Response(200, json=card) + return httpx.Response(200, content=_streamed()) + + transport = httpx.MockTransport(site) + monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw)) + gw = _gateway(_Changelog(), website_urls={"shop": "http://shop-website-backend:8000"}, + rest_proxy_urls={"shop": "http://shop-website-backend:8000"}) + for path in ("mcp", "mcp/", "./mcp"): + assert (await _http(gw, "POST", f"/website/shop/{path}", body=b"{}"))[0] == 404, path + assert (await _http(gw, "GET", "/website/shop/orders"))[0] == 200 + @pytest.mark.asyncio async def test_a_sites_route_refuses_what_its_card_advertises_reading_the_card_once(monkeypatch): reached = []