Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions packages/agentenv-protocol/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,16 @@ with any other transport are kept beside the SDK's entry. Card URLs are paths, r
address the card was fetched from. On the client side, `client.mcp_path(card)` returns the
declared path, or `/mcp` for a card without one.

A composed card, such as a gateway's, nests each backing environment's card under
`children_environments`, its `url` and endpoints rooted at the route the composer serves it at.
`client.children(card)` returns each as a `ChildEnv` (its `name`, `url` and `card`, with its
`extensions`, `operations` and `serves(method)`, and `base_url(address)`, its data plane resolved
against the composed card's address). Two children may share a name, as a website and an MCP
server can, so `client.child_at(card, url)` finds one by its url or route; `client.child_named(card,
name)` raises `AmbiguousChildError` for a shared name, and `client.find_child` returns the first.
`client.serves(card, method)` says whether a card serves a data-plane method by its declared
`operations`, None for a card that declares none.

Extensions are invoked from what the card advertises. `client.find_extension_method(card, uri,
method)` returns one advertised method, whose `endpoint` is the method's own or else the
extension's, and `client.invoke_extension(base_url, card, uri, params, method=...)` calls it with
Expand Down
80 changes: 80 additions & 0 deletions packages/agentenv-protocol/src/agentenv_protocol/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import logging
from collections.abc import Iterator
from contextvars import ContextVar
from dataclasses import dataclass, field

import httpx

Expand Down Expand Up @@ -210,6 +211,83 @@ def find_tool(card: dict, name: str) -> dict | None:
return None


def serves(card: dict, method: str) -> bool | None:
"""Whether a card serves the data-plane ``method`` (``data/get`` say): True or False by its declared
``capabilities.operations``, or None when it declares none, as an older card doesn't, so the caller decides as it
did before cards declared them (a probe, say)."""
declared = operations(card)
return None if declared is None else method in declared


@dataclass(frozen=True)
class ChildEnv:
"""One child env a composed card nests under ``children_environments``: its ``name`` and ``url`` as its card gives
them, and that ``card``, which says what it offers. A composer roots each child's ``url`` (and its endpoints) at
the route it serves the child at, so two children sharing a name have distinct urls; ``url`` is None for a child
whose card names none."""

name: str | None
url: str | None
card: dict = field(hash=False, repr=False)

@property
def extensions(self) -> list[dict]:
"""The extensions the child's card advertises (``capabilities.extensions``)."""
return list((self.card.get("capabilities") or {}).get("extensions") or [])

@property
def operations(self) -> list[str] | None:
"""The data-plane methods the child's card declares (``operations``), or None when it declares none."""
return operations(self.card)

def find_extension(self, uri: str) -> dict | None:
"""The extension the child's card advertises under ``uri``, or None (``find_extension``)."""
return find_extension(self.card, uri)

def serves(self, method: str) -> bool | None:
"""Whether the child's card serves the data-plane ``method``, None when it declares no operations (``serves``)."""
return serves(self.card, method)

def base_url(self, address: str) -> str:
"""Where the child's data plane is, its ``url`` without the RPC path, resolved against ``address``, the composed
card's: a path joins onto it, a full URL stands alone, and a child naming no url is at ``address`` itself."""
url = (self.url or RPC_PATH).removesuffix(RPC_PATH)
return url if "://" in url else f"{address.rstrip('/')}{url}"


def children(card: dict) -> list[ChildEnv]:
"""Each child env a composed card nests under ``children_environments``, in its order; none for a leaf card, or
one whose ``children_environments`` is missing or null. An entry that isn't an object is skipped."""
return [ChildEnv(name=child.get("name"), url=child.get("url"), card=child)
for child in card.get("children_environments") or [] if isinstance(child, dict)]


def child_at(card: dict, url: str) -> ChildEnv | None:
"""The child env of a composed card at ``url``: the one whose ``url`` is ``url``, or whose data plane is there
(``url`` without the RPC path, ``/svc/mcp-slack`` for ``/svc/mcp-slack/agentenv`` say), a trailing slash aside;
None when no child is. A composer serves each child at its own route, so this is the lookup two children sharing a
name can't confuse."""
wanted = url.rstrip("/")
for child in children(card):
if child.url is not None and wanted in (child.url.rstrip("/"), child.url.removesuffix(RPC_PATH).rstrip("/")):
return child
return None


class AmbiguousChildError(LookupError):
"""More than one child env of a composed card has the name asked for, so the name can't say which is meant."""


def child_named(card: dict, name: str) -> ChildEnv | None:
"""The one child env of a composed card named ``name``, or None when none is; raises ``AmbiguousChildError`` when
several are, as a website and an MCP server sharing a name are, which ``child_at`` tells apart by their urls."""
found = [child for child in children(card) if child.name == name]
if len(found) > 1:
raise AmbiguousChildError(f"{len(found)} child envs are named {name!r} ({', '.join(str(c.url) for c in found)}); "
"look one up by its url")
return found[0] if found else None


def find_child(card: dict, name: str) -> dict | None:
"""Return the nested child EnvironmentCard named `name`, or None.

Expand All @@ -218,6 +296,8 @@ def find_child(card: dict, name: str) -> dict | None:
rewritten to be reachable from the composed card's base_url. Callers resolve a service by
name here, then use the child card with `find_extension`/`invoke_extension`. Tolerates a
missing/null `children_environments` so single/leaf cards return None rather than raising.
Two children sharing `name` return the first; `child_named` refuses that, and `child_at`
finds a child by its url, which tells them apart.
"""
for child in card.get("children_environments") or []:
if child.get("name") == name:
Expand Down
52 changes: 52 additions & 0 deletions packages/agentenv-protocol/tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,58 @@ def test_find_child_tolerates_missing_children():
assert protocol_v1.find_child({"children_environments": None}, "slack") is None


_SHARED_NAME = {"name": "gw", "children_environments": [
{"name": "mail", "url": "/svc/mcp-mail/agentenv", "capabilities": {"operations": ["data/get"], "extensions": [
{"uri": "urn:agentenv:clock/v1"}]}},
{"name": "mail", "url": "/svc/mail/agentenv", "capabilities": {"operations": []}},
{"name": "docs", "capabilities": {}},
"not a card",
]}


def test_children_are_each_nested_card_by_its_name_url_and_card():
found = protocol_v1.children(_SHARED_NAME)

assert [(c.name, c.url) for c in found] == [("mail", "/svc/mcp-mail/agentenv"), ("mail", "/svc/mail/agentenv"), ("docs", None)]
assert found[0].card is _SHARED_NAME["children_environments"][0]
assert [c.extensions for c in found] == [[{"uri": "urn:agentenv:clock/v1"}], [], []]
assert [c.operations for c in found] == [["data/get"], [], None]
assert [c.serves("data/get") for c in found] == [True, False, None]
assert found[0].find_extension("urn:agentenv:clock/v1") is not None and found[1].find_extension("urn:agentenv:clock/v1") is None
assert protocol_v1.children({"name": "leaf"}) == [] and protocol_v1.children({"children_environments": None}) == []


def test_a_childs_base_url_is_its_url_without_the_rpc_path_against_the_composed_cards_address():
mcp, website, docs = protocol_v1.children(_SHARED_NAME)

assert mcp.base_url("https://env.example/") == "https://env.example/svc/mcp-mail"
assert website.base_url("https://env.example") == "https://env.example/svc/mail"
assert docs.base_url("https://env.example") == "https://env.example"
assert protocol_v1.ChildEnv("x", "https://x.example/agentenv", {}).base_url("https://env.example") == "https://x.example"


def test_child_at_finds_a_child_by_its_url_or_data_plane_route_where_two_share_a_name():
assert protocol_v1.child_at(_SHARED_NAME, "/svc/mail").url == "/svc/mail/agentenv"
assert protocol_v1.child_at(_SHARED_NAME, "/svc/mcp-mail/").url == "/svc/mcp-mail/agentenv"
assert protocol_v1.child_at(_SHARED_NAME, "/svc/mail/agentenv").url == "/svc/mail/agentenv"
assert protocol_v1.child_at(_SHARED_NAME, "/svc/docs") is None
assert protocol_v1.child_at({"name": "leaf"}, "/svc/mail") is None


def test_child_named_refuses_a_name_two_children_share():
assert protocol_v1.child_named(_SHARED_NAME, "docs").name == "docs"
assert protocol_v1.child_named(_SHARED_NAME, "gmail") is None
with pytest.raises(protocol_v1.AmbiguousChildError, match="2 child envs are named 'mail'"):
protocol_v1.child_named(_SHARED_NAME, "mail")
assert protocol_v1.find_child(_SHARED_NAME, "mail")["url"] == "/svc/mcp-mail/agentenv"


def test_serves_is_what_a_card_declares_of_a_method_and_none_when_it_declares_no_operations():
assert protocol_v1.serves({"capabilities": {"operations": ["data/get"]}}, "data/get") is True
assert protocol_v1.serves({"capabilities": {"operations": ["data/add"]}}, "data/get") is False
assert protocol_v1.serves({"capabilities": {}}, "data/get") is None


def test_find_extension_ignores_children():
# set-errors/v1 is advertised only on the children, never at the top level
assert protocol_v1.find_extension(_CARD_WITH_CHILDREN, "urn:agentenv:set-errors/v1") is None
Expand Down
128 changes: 122 additions & 6 deletions src/agent_env/env/legacy_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,15 @@
from __future__ import annotations

import asyncio
import dataclasses
import logging
from typing import TYPE_CHECKING, Optional
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Optional

import httpx
from agentenv_protocol import RPC_PATH, DataPart, client as protocol_v1
from agentenv_protocol import METHOD_GET, RPC_PATH, DataPart, client as protocol_v1

from agent_env.env.gateway.constants import EXT_WEBSITE_URI

if TYPE_CHECKING:
from agent_env.env.env import DeployedEnv
Expand All @@ -33,20 +37,132 @@ def environment_base_url(gateway_url: Optional[str], environment_name: str, mcp:
return f"{gateway_url}/svc/{prefix}{environment_name}"


@dataclass(frozen=True)
class BackingServer:
"""A server behind an env's address, by its environment_name: an MCP server or, with ``mcp=False``, a website.

One the env's card lists (``card_servers``) carries that ``child`` and the ``address`` its paths resolve against.
One the card lists none for, a server serving no card, carries the ``address`` alone: it has no card, and none is
read for it. One named by the env definition of a record whose card can't be read carries neither, so its card is
read live at its gateway path, ``/svc/mcp-{name}`` or a website's ``/svc/{name}`` (``server_card``)."""

environment_name: str
mcp: bool = True
child: Optional[protocol_v1.ChildEnv] = None
address: Optional[str] = None

@property
def card(self) -> Optional[dict]:
"""The server's card as the env's card lists it, or None."""
return self.child.card if self.child is not None else None


def definition_servers(env: Any) -> list[BackingServer]:
"""The servers an env definition names: a MultiEnv's MCP servers then its websites, a WebsiteEnv's website, or
else the env's one MCP server; empty for an env with none. Read only for what the env's card can't say
(``backing_servers``)."""
from agent_env.env.envs.website import WebsiteEnv

if isinstance(env, WebsiteEnv):
return [BackingServer(env.environment_name, mcp=False)]
servers = [BackingServer(e.environment_name) for e in getattr(env, "mcp_server_envs", None) or []]
servers += [BackingServer(e.environment_name, mcp=False) for e in getattr(env, "website_envs", None) or []]
if not servers and getattr(env, "environment_name", None):
servers = [BackingServer(env.environment_name)]
return servers


def card_servers(card: dict, address: str) -> list[BackingServer]:
"""The servers ``card``, read at ``address``, lists, each with its own card: a leaf card's env itself; else each
child the composed card nests, after the env itself when its own card serves ``data/get``, as a computer's driver
fronting its apps does. A child is a website when its card carries ``website/v1`` or its url is at a website's
gateway route, ``/svc/{name}`` (an MCP server's is ``/svc/mcp-{name}``), so a website and an MCP server sharing a
name are two servers."""
own = protocol_v1.ChildEnv(name=card.get("name"), url=card.get("url"), card=card)
if not isinstance(card.get("children_environments"), list):
return [_listed(own, address)]
servers = [_listed(own, address)] if protocol_v1.serves(card, METHOD_GET) else []
return servers + [_listed(child, address) for child in protocol_v1.children(card)]


def _listed(child: protocol_v1.ChildEnv, address: str) -> BackingServer:
route, url = f"/svc/{child.name}", (child.url or "").rstrip("/")
website = child.find_extension(EXT_WEBSITE_URI) is not None or url == route or url.startswith(f"{route}/")
return BackingServer(child.name, mcp=not website, child=child, address=address)


async def backing_servers(deployed: Optional[DeployedEnv], gateway_url: Optional[str], env: Any,
timeout: int = 10) -> list[BackingServer]:
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
"""The servers behind an env as its card lists them (``card_servers``), then, behind a gateway, each server its
definition ``env`` names that the card lists none of the same name and kind for: one serving no card, which the
gateway leaves off its card, so it has none.

The card is ``deployed``'s stored one when ``gateway_url`` is its own gateway or None, else read live at
``gateway_url`` (an override, or a record that kept no card). A record with no card URL, an older one, or no
record, or a live read that fails, leaves the definition alone: its servers, each read live at its gateway path,
logged once."""
from agent_env.env.env import gateway_url_of

gateway = gateway_url.rstrip("/") if gateway_url else None
card, address, why = None, None, "it has no record"
if deployed is not None and deployed.environment_card and (gateway is None or (gateway_url_of(deployed) or "").rstrip("/") == gateway):
card, address = deployed.environment_card, deployed.environment_url
elif deployed is not None and deployed.environment_card_url and gateway:
try:
card, address = await protocol_v1.get_card(gateway, timeout=timeout), gateway
except httpx.HTTPError as e:
why = f"its card at {gateway} could not be read ({type(e).__name__})"
elif deployed is not None:
why = "its record has no card URL"
named = definition_servers(env)
if not isinstance(card, dict):
logger.info(f"env {getattr(env, 'id', None)!r}: {why}, so its servers are the ones its definition names, "
f"each read live at its gateway path: {[(s.environment_name, s.mcp) for s in named]}")
return named
servers = card_servers(card, address)
if not gateway:
return servers
listed = {(s.environment_name, s.mcp) for s in servers}
return servers + [dataclasses.replace(s, address=address) for s in named if (s.environment_name, s.mcp) not in listed]


async def server_card(server: BackingServer, gateway_url: Optional[str], timeout: int = 10) -> tuple[Optional[str], Optional[dict]]:
"""A server's card and the base its endpoints resolve against: as the env's card lists it, None for a server the
card lists none for, else read live at its gateway path (``child_env_card``)."""
if server.address is not None:
return server.address, server.card
return await child_env_card(None, gateway_url, server.environment_name, timeout=timeout, mcp=server.mcp)


async def server_v1_base_url(server: BackingServer, gateway_url: Optional[str]) -> Optional[str]:
"""A server's v1 data-plane base: its card's ``url`` resolved against the env's address, None for a server the
env's card lists none for, else as the live probe at its gateway path finds it (``v1_base_url``)."""
if server.address is not None:
return server.child.base_url(server.address) if server.child is not None else None
return await v1_base_url(None, gateway_url, server.environment_name, mcp=server.mcp)


async def v1_base_url(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, mcp: bool = True) -> Optional[str]:
"""A child env's v1 data-plane base from the stored env card, or None for legacy; without a stored card, the live probe decides."""
"""A child env's v1 data-plane base from the stored env card, or None for legacy; without a stored card, the live probe
decides. A website's probe is a card read that is None only on a 404 and raises on any other failure, so a website whose
card could not be read is never taken for one serving no data plane."""
if deployed is None or not deployed.environment_card:
base_url = environment_base_url(gateway_url, environment_name, mcp=mcp)
if not mcp:
_, card = await child_env_card(None, gateway_url, environment_name, mcp=False)
return base_url if card is not None else None
return base_url if await protocol_v1.supports_v1(base_url) else None
child = deployed.get_child_env_card(environment_name)
return deployed.environment_url + child.get("url", RPC_PATH).removesuffix(RPC_PATH) if child else None



async def child_env_card(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, timeout: int = 10) -> tuple[str, Optional[dict]]:
"""An MCP child env's card and the base its endpoints resolve against: from the stored env card, or without one a live read (None on 404)."""
async def child_env_card(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, timeout: int = 10,
mcp: bool = True) -> tuple[str, Optional[dict]]:
"""A child env's card and the base its endpoints resolve against: from the stored env card, or without one a live read
at the MCP server's gateway path, or with ``mcp=False`` the website's (None on 404)."""
if deployed is None or not deployed.environment_card:
base_url = environment_base_url((gateway_url or "").rstrip("/"), environment_name)
base_url = environment_base_url((gateway_url or "").rstrip("/"), environment_name, mcp=mcp)
try:
return base_url, await protocol_v1.get_card(base_url, timeout=timeout)
except httpx.HTTPStatusError as e:
Expand Down
Loading
Loading