fix(adagents): normalize agent matching and expose listing lookup (#1329)

Author: bokelley

docs/adagents-discovery.md

@@ -0,0 +1,22 @@
+# Publisher agent discovery
+
+Agent lookup treats host case and explicit default ports as equivalent, and
+continues to match HTTP and HTTPS URLs. Paths remain case-sensitive; trailing
+slashes, query strings and fragments do not affect matching. Non-default ports
+remain significant.
+
+Use the listing helper alongside property resolution to distinguish an agent
+that is listed but whose selectors match no properties from an unlisted agent:
+
+```python
+from adcp import find_authorized_agent_entries, get_properties_by_agent
+
+entries = find_authorized_agent_entries(document, "https://sales.example.com/mcp")
+properties = get_properties_by_agent(document, "https://sales.example.com/mcp")
+listed_but_unbound = bool(entries) and not properties
+```
+
+Malformed sibling entries and non-string or empty URL values are skipped during
+lookup. `validate_adagents_structure(document)` reports invalid URL types through
+its existing `missing_url` diagnostic. The listing helper returns matching entry
+objects in source order and does not modify the document.

src/adcp/__init__.py

@@ -158,6 +158,7 @@ def _resolve_version() -> str:
         "fetch_agent_authorizations",
         "fetch_agent_authorizations_from_directory",
         "filter_revoked_selectors",
+        "find_authorized_agent_entries",
         "get_all_properties",
         "get_all_tags",
         "get_properties_by_agent",
@@ -1555,6 +1556,7 @@ def get_adcp_version() -> str:
     "fetch_agent_authorizations",
     "fetch_agent_authorizations_from_directory",
     "filter_revoked_selectors",
+    "find_authorized_agent_entries",
     "validate_adagents_domain",
     "validate_adagents_structure",
     "verify_agent_authorization",
@@ -1803,6 +1805,7 @@ def get_adcp_version() -> str:
         fetch_agent_authorizations,
         fetch_agent_authorizations_from_directory,
         filter_revoked_selectors,
+        find_authorized_agent_entries,
         get_all_properties,
         get_all_tags,
         get_properties_by_agent,

src/adcp/adagents.py

@@ -456,17 +456,60 @@ def _resolve_well_known_redirect_url(
 
 
 def normalize_url(url: str) -> str:
-    """Normalize URL by removing protocol and trailing slash.
+    """Return a protocol-agnostic URL key with host case/default ports normalized.
 
-    Args:
-        url: URL to normalize
-
-    Returns:
-        Normalized URL (domain/path without protocol or trailing slash)
+    Paths remain case-sensitive. Query strings and fragments are ignored, and
+    trailing path slashes are removed, preserving the existing matching rules.
+    Only the default port for the supplied scheme is removed (80 for HTTP,
+    443 for HTTPS); non-default ports remain significant.
     """
     parsed = urlparse(url)
-    normalized = parsed.netloc + parsed.path
-    return normalized.rstrip("/")
+    authority = parsed.netloc
+    if parsed.hostname is not None:
+        host = parsed.hostname.lower()
+        if ":" in host:
+            host = f"[{host}]"
+        port = parsed.port
+        if port is not None and (parsed.scheme.lower(), port) not in {("http", 80), ("https", 443)}:
+            host += f":{port}"
+        userinfo = authority.rsplit("@", 1)[0] + "@" if "@" in authority else ""
+        authority = userinfo + host
+    return (authority + parsed.path).rstrip("/")
+
+
+def find_authorized_agent_entries(
+    adagents_data: dict[str, Any], agent_url: str
+) -> list[dict[str, Any]]:
+    """Return all entries listing an agent, independently of property binding.
+
+    A non-empty result distinguishes a listed agent whose selectors resolve to
+    no properties from an unlisted agent. Entries are returned in document order
+    without copying or modifying them. Malformed sibling entries and URL values
+    are skipped. Matching preserves HTTP/HTTPS equivalence and path semantics.
+
+    Raises AdagentsValidationError if the document is not an object with an
+    ``authorized_agents`` array.
+    """
+    if not isinstance(adagents_data, dict):
+        raise AdagentsValidationError("adagents_data must be a dictionary")
+    authorized_agents = adagents_data.get("authorized_agents")
+    if not isinstance(authorized_agents, list):
+        raise AdagentsValidationError("adagents.json must have 'authorized_agents' array")
+    normalized_agent_url = normalize_url(agent_url)
+    matches: list[dict[str, Any]] = []
+    for agent in authorized_agents:
+        if not isinstance(agent, dict):
+            continue
+        entry_url = agent.get("url")
+        if not isinstance(entry_url, str) or not entry_url.strip():
+            continue
+        try:
+            normalized_entry_url = normalize_url(entry_url)
+        except ValueError:
+            continue
+        if normalized_entry_url == normalized_agent_url:
+            matches.append(agent)
+    return matches
 
 
 def domain_matches(property_domain: str, agent_domain_pattern: str) -> bool:
@@ -584,30 +627,7 @@ def verify_agent_authorization(
         - Implements AdCP domain matching rules
         - Agent URLs are matched ignoring protocol and trailing slash
     """
-    # Validate structure
-    if not isinstance(adagents_data, dict):
-        raise AdagentsValidationError("adagents_data must be a dictionary")
-
-    authorized_agents = adagents_data.get("authorized_agents")
-    if not isinstance(authorized_agents, list):
-        raise AdagentsValidationError("adagents.json must have 'authorized_agents' array")
-
-    # Normalize the agent URL for comparison
-    normalized_agent_url = normalize_url(agent_url)
-
-    # Check each authorized agent
-    for agent in authorized_agents:
-        if not isinstance(agent, dict):
-            continue
-
-        agent_url_from_json = agent.get("url", "")
-        if not agent_url_from_json:
-            continue
-
-        # Match agent URL (protocol-agnostic)
-        if normalize_url(agent_url_from_json) != normalized_agent_url:
-            continue
-
+    for agent in find_authorized_agent_entries(adagents_data, agent_url):
         # Found matching agent - now check properties
         properties = agent.get("properties")
 
@@ -1915,12 +1935,7 @@ def _resolve_properties_for_agent(
     permissive_bare_top_level: bool,
 ) -> list[dict[str, Any]]:
     """Implementation shared by strict and opt-in permissive property resolution."""
-    if not isinstance(adagents_data, dict):
-        raise AdagentsValidationError("adagents_data must be a dictionary")
-
-    authorized_agents = adagents_data.get("authorized_agents")
-    if not isinstance(authorized_agents, list):
-        raise AdagentsValidationError("adagents.json must have 'authorized_agents' array")
+    matches = find_authorized_agent_entries(adagents_data, agent_url)
 
     top_level_properties = adagents_data.get("properties", [])
     if not isinstance(top_level_properties, list):
@@ -1937,23 +1952,8 @@ def _resolve_properties_for_agent(
         )
     ]
 
-    normalized_agent_url = normalize_url(agent_url)
-
     domain_index = _build_domain_index(revoked_top_level)
 
-    matches: list[dict[str, Any]] = []
-    for agent in authorized_agents:
-        if not isinstance(agent, dict):
-            continue
-
-        agent_url_from_json = agent.get("url", "")
-        if not agent_url_from_json:
-            continue
-
-        if normalize_url(agent_url_from_json) != normalized_agent_url:
-            continue
-        matches.append(agent)
-
     resolved: list[dict[str, Any]] = []
     for agent in matches:
         # revoked_top_level pre-filters revoked domains from the per-domain

tests/fixtures/public_api_snapshot.json

@@ -678,6 +678,7 @@
     "fetch_agent_authorizations",
     "fetch_agent_authorizations_from_directory",
     "filter_revoked_selectors",
+    "find_authorized_agent_entries",
     "format_is_supported",
     "formats_are_equivalent",
     "generate_webhook_challenge_value",

tests/test_adagents_matching.py

@@ -0,0 +1,100 @@
+"""Regression contracts for protocol-agnostic agent listing and resolution."""
+
+from copy import deepcopy
+from typing import Any
+
+import pytest
+
+from adcp import find_authorized_agent_entries
+from adcp.adagents import (
+    get_properties_by_agent,
+    normalize_url,
+    resolve_properties_for_agent,
+    validate_adagents_structure,
+    verify_agent_authorization,
+)
+
+AGENT = "https://sales.example.com/mcp"
+
+
+def document(url: Any = AGENT, tags: list[str] | None = None) -> dict[str, Any]:
+    return {
+        "authorized_agents": [
+            {
+                "url": url,
+                "authorized_for": "Display",
+                "authorization_type": "property_tags",
+                "property_tags": ["all"] if tags is None else tags,
+            }
+        ],
+        "properties": [
+            {
+                "property_id": "site",
+                "property_type": "website",
+                "name": "Site",
+                "identifiers": [{"type": "domain", "value": "example.com"}],
+                "tags": ["all"],
+            }
+        ],
+    }
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://SALES.Example.com:443/mcp/",
+        "http://sales.example.com:80/mcp",
+        "https://sales.example.com/mcp/?q=x#fragment",
+    ],
+)
+def test_normalization_retains_protocol_and_path_contract(url: str) -> None:
+    assert normalize_url(url) == "sales.example.com/mcp"
+    doc = document(url)
+    assert find_authorized_agent_entries(doc, AGENT) == doc["authorized_agents"]
+    assert get_properties_by_agent(doc, AGENT) == doc["properties"]
+    assert verify_agent_authorization(doc, AGENT)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://sales.example.com:80/mcp",
+        "http://sales.example.com:443/mcp",
+        "https://sales.example.com:8443/mcp",
+        "https://sales.example.com/MCP",
+        "https://sales.example.com/other",
+    ],
+)
+def test_paths_and_nondefault_ports_remain_significant(url: str) -> None:
+    assert find_authorized_agent_entries(document(url), AGENT) == []
+
+
+@pytest.mark.parametrize(
+    "url",
+    [None, [], [AGENT], 42, {}, "", "   ", "https://sales.example.com:bad/mcp", "https://[bad/mcp"],
+)
+def test_invalid_sibling_cannot_break_lookup(url: Any) -> None:
+    doc = document(url)
+    valid = document()["authorized_agents"][0]
+    doc["authorized_agents"].extend([None, 1, valid])
+    before = deepcopy(doc)
+    assert find_authorized_agent_entries(doc, AGENT) == [valid]
+    assert verify_agent_authorization(doc, AGENT)
+    assert get_properties_by_agent(doc, AGENT) == doc["properties"]
+    assert resolve_properties_for_agent(doc, AGENT, mode="permissive") == doc["properties"]
+    assert doc == before
+    if not isinstance(url, str) or not url:
+        assert any(error.kind == "missing_url" for error in validate_adagents_structure(doc).errors)
+
+
+def test_listing_distinguishes_unbound_and_unlisted_and_retains_duplicates() -> None:
+    unbound = document(tags=["missing"])
+    assert get_properties_by_agent(unbound, AGENT) == []
+    assert find_authorized_agent_entries(unbound, AGENT)
+    unlisted = document("https://other.example.com/mcp")
+    assert get_properties_by_agent(unlisted, AGENT) == []
+    assert find_authorized_agent_entries(unlisted, AGENT) == []
+    unbound["authorized_agents"].append(document()["authorized_agents"][0])
+    matches = find_authorized_agent_entries(unbound, AGENT)
+    assert matches == unbound["authorized_agents"]
+    assert matches[0] is unbound["authorized_agents"][0]

tests/type_checks/adagents_matching.py

@@ -0,0 +1,10 @@
+"""The public listing helper retains typed JSON entries for adopter code."""
+
+from typing import Any
+
+from adcp import find_authorized_agent_entries
+
+
+def is_listed(document: dict[str, Any], agent_url: str) -> bool:
+    entries: list[dict[str, Any]] = find_authorized_agent_entries(document, agent_url)
+    return bool(entries)