feat: Add publisher authorization discovery API (#54)

bokelley · claude · web-flow · commit e7d06963c7d9 · 2025-11-16T19:40:14.000-05:00
* feat: Add publisher authorization discovery API Implement two approaches for discovering which publishers have authorized an agent: 1. "Pull" approach: fetch_agent_authorizations() - Check publisher adagents.json files in parallel to see which ones authorize your agent, extracting property IDs and tags. 2. "Push" approach: ADCPClient.list_authorized_properties() - Ask the agent directly what publishers it represents (already existed, now documented). Add AuthorizationContext class to encapsulate authorization info (property IDs, tags, raw property data). Includes comprehensive tests covering success, failure, filtering, and connection pooling scenarios. Resolves #53 🤖 Generated with Claude Code Co-Authored-By: Claude <noreply@anthropic.com> * docs: Move authorization discovery to README Remove separate AUTHORIZATION_DISCOVERY.md file and integrate content into README as a subsection under Publisher Authorization Validation. Keeps documentation consolidated in a single location. 🤖 Generated with Claude Code Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com>
diff --git a/README.md b/README.md
@@ -505,6 +505,42 @@ is_authorized = await verify_agent_for_property(
 
 See `examples/adagents_validation.py` for complete examples.
 
+### Authorization Discovery
+
+Discover which publishers have authorized your agent using two approaches:
+
+**1. "Push" Approach** - Ask the agent (recommended, fastest):
+```python
+from adcp import ADCPClient
+
+async with ADCPClient(agent_config) as client:
+    # Single API call to agent
+    response = await client.simple.list_authorized_properties()
+    print(f"Authorized for: {response.publisher_domains}")
+```
+
+**2. "Pull" Approach** - Check publisher adagents.json files (when you need property details):
+```python
+from adcp import fetch_agent_authorizations
+
+# Check specific publishers (fetches in parallel)
+contexts = await fetch_agent_authorizations(
+    "https://our-sales-agent.com",
+    ["nytimes.com", "wsj.com", "cnn.com"]
+)
+
+for domain, ctx in contexts.items():
+    print(f"{domain}:")
+    print(f"  Property IDs: {ctx.property_ids}")
+    print(f"  Tags: {ctx.property_tags}")
+```
+
+**When to use which:**
+- **Push**: Quick discovery, portfolio overview, high-level authorization check
+- **Pull**: Property-level details, specific publisher list, works offline
+
+See `examples/fetch_agent_authorizations.py` for complete examples.
+
 ## CLI Tool
 
 The `adcp` command-line tool provides easy interaction with AdCP agents without writing code.
diff --git a/examples/fetch_agent_authorizations.py b/examples/fetch_agent_authorizations.py
@@ -0,0 +1,144 @@
+"""
+Example showing how to discover which publishers have authorized your agent.
+
+This example demonstrates TWO approaches:
+
+1. "Push" approach - Ask the agent what it's authorized for:
+   - Use the agent's list_authorized_properties endpoint
+   - Agent tells you which publisher domains it represents
+   - Fast and efficient - single API call
+
+2. "Pull" approach - Check publisher adagents.json files:
+   - Use fetch_agent_authorizations to check multiple publishers
+   - Fetch adagents.json from each publisher's .well-known directory
+   - Useful when you have a specific list of publishers to check
+   - Supports connection pooling for better performance
+"""
+
+import asyncio
+
+from adcp import ADCPClient, AgentConfig, Protocol, fetch_agent_authorizations
+
+
+async def approach_1_push():
+    """APPROACH 1: Ask the agent what it's authorized for (RECOMMENDED)."""
+    print("=" * 70)
+    print("APPROACH 1: Push - Ask agent what it's authorized for")
+    print("=" * 70)
+    print()
+
+    # Configure the agent client
+    agent_config = AgentConfig(
+        id="sales_agent",
+        agent_uri="https://our-sales-agent.com",
+        protocol=Protocol.A2A,
+    )
+
+    async with ADCPClient(agent_config) as client:
+        # Ask the agent directly what publishers it represents
+        # This is fast - just one API call!
+        response = await client.simple.list_authorized_properties()
+
+        print(f"✅ Agent represents {len(response.publisher_domains)} publishers:\n")
+
+        for domain in response.publisher_domains:
+            print(f"  • {domain}")
+
+        print()
+        print("📊 Portfolio Summary:")
+        if response.primary_channels:
+            print(f"  Primary Channels: {', '.join(response.primary_channels)}")
+        if response.primary_countries:
+            print(f"  Primary Countries: {', '.join(response.primary_countries)}")
+        if response.portfolio_description:
+            print(f"  Description: {response.portfolio_description[:100]}...")
+
+        print()
+        print("💡 TIP: Now fetch each publisher's adagents.json to see property details")
+        print()
+
+
+async def approach_2_pull():
+    """APPROACH 2: Check publisher adagents.json files (when you know which publishers to check)."""
+    print("=" * 70)
+    print("APPROACH 2: Pull - Check specific publisher adagents.json files")
+    print("=" * 70)
+    print()
+
+    # Your agent's URL
+    agent_url = "https://our-sales-agent.com"
+
+    # Publisher domains to check
+    publisher_domains = [
+        "nytimes.com",
+        "wsj.com",
+        "cnn.com",
+        "espn.com",
+        "techcrunch.com",
+    ]
+
+    print(f"Checking authorization for {agent_url} across {len(publisher_domains)} publishers...\n")
+
+    # Fetch authorization contexts (fetches all in parallel)
+    contexts = await fetch_agent_authorizations(agent_url, publisher_domains)
+
+    # Display results
+    if not contexts:
+        print("No authorizations found.")
+        return
+
+    print(f"✅ Authorized for {len(contexts)}/{len(publisher_domains)} publishers:\n")
+
+    for domain, ctx in contexts.items():
+        print(f"{domain}:")
+        print(f"  Property IDs: {ctx.property_ids}")
+        print(f"  Property Tags: {ctx.property_tags}")
+        print(f"  Total Properties: {len(ctx.raw_properties)}")
+        print()
+
+    # Example: Check if specific tags are available
+    all_tags = set()
+    for ctx in contexts.values():
+        all_tags.update(ctx.property_tags)
+
+    print(f"📊 Total unique tags across all publishers: {len(all_tags)}")
+    print(f"Tags: {sorted(all_tags)}")
+    print()
+
+
+async def main():
+    """Demonstrate both approaches."""
+    # APPROACH 1: Fast - ask agent what it's authorized for
+    await approach_1_push()
+
+    print("\n" + "=" * 70 + "\n")
+
+    # APPROACH 2: Check specific publishers
+    await approach_2_pull()
+
+
+async def main_with_connection_pooling():
+    """More efficient version using connection pooling for multiple requests."""
+    import httpx
+
+    agent_url = "https://our-sales-agent.com"
+    publisher_domains = ["nytimes.com", "wsj.com", "cnn.com"]
+
+    # Use a shared HTTP client for connection pooling
+    async with httpx.AsyncClient(
+        limits=httpx.Limits(max_keepalive_connections=10, max_connections=20)
+    ) as client:
+        print("Using connection pooling for better performance...\n")
+
+        contexts = await fetch_agent_authorizations(agent_url, publisher_domains, client=client)
+
+        for domain, ctx in contexts.items():
+            print(f"{domain}: {len(ctx.property_ids)} properties")
+
+
+if __name__ == "__main__":
+    # Run basic example
+    asyncio.run(main())
+
+    # Uncomment to run connection pooling example
+    # asyncio.run(main_with_connection_pooling())
diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py
@@ -8,8 +8,10 @@
 """
 
 from adcp.adagents import (
+    AuthorizationContext,
     domain_matches,
     fetch_adagents,
+    fetch_agent_authorizations,
     get_all_properties,
     get_all_tags,
     get_properties_by_agent,
@@ -178,7 +180,9 @@
     "Product",
     "Property",
     # Adagents validation
+    "AuthorizationContext",
     "fetch_adagents",
+    "fetch_agent_authorizations",
     "verify_agent_authorization",
     "verify_agent_for_property",
     "domain_matches",
diff --git a/src/adcp/adagents.py b/src/adcp/adagents.py
@@ -518,3 +518,125 @@ def get_properties_by_agent(adagents_data: dict[str, Any], agent_url: str) -> li
         return [p for p in properties if isinstance(p, dict)]
 
     return []
+
+
+class AuthorizationContext:
+    """Authorization context for a publisher domain.
+
+    Attributes:
+        property_ids: List of property IDs the agent is authorized for
+        property_tags: List of property tags the agent is authorized for
+        raw_properties: Raw property data from adagents.json
+    """
+
+    def __init__(self, properties: list[dict[str, Any]]):
+        """Initialize from list of properties.
+
+        Args:
+            properties: List of property dictionaries from adagents.json
+        """
+        self.property_ids: list[str] = []
+        self.property_tags: list[str] = []
+        self.raw_properties = properties
+
+        # Extract property IDs and tags
+        for prop in properties:
+            if not isinstance(prop, dict):
+                continue
+
+            # Extract property ID
+            prop_id = prop.get("id")
+            if prop_id and isinstance(prop_id, str):
+                self.property_ids.append(prop_id)
+
+            # Extract tags
+            tags = prop.get("tags", [])
+            if isinstance(tags, list):
+                for tag in tags:
+                    if isinstance(tag, str) and tag not in self.property_tags:
+                        self.property_tags.append(tag)
+
+    def __repr__(self) -> str:
+        return (
+            f"AuthorizationContext("
+            f"property_ids={self.property_ids}, "
+            f"property_tags={self.property_tags})"
+        )
+
+
+async def fetch_agent_authorizations(
+    agent_url: str,
+    publisher_domains: list[str],
+    timeout: float = 10.0,
+    client: httpx.AsyncClient | None = None,
+) -> dict[str, AuthorizationContext]:
+    """Fetch authorization contexts by checking publisher adagents.json files.
+
+    This function discovers what publishers have authorized your agent by fetching
+    their adagents.json files from the .well-known directory and extracting the
+    properties your agent can access.
+
+    This is the "pull" approach - you query publishers to see if they've authorized you.
+    For the "push" approach where the agent tells you what it's authorized for,
+    use the agent's list_authorized_properties endpoint via ADCPClient.
+
+    Args:
+        agent_url: URL of your sales agent
+        publisher_domains: List of publisher domains to check (e.g., ["nytimes.com", "wsj.com"])
+        timeout: Request timeout in seconds for each fetch
+        client: Optional httpx.AsyncClient for connection pooling
+
+    Returns:
+        Dictionary mapping publisher domain to AuthorizationContext.
+        Only includes domains where the agent is authorized.
+
+    Example:
+        >>> # "Pull" approach - check what publishers have authorized you
+        >>> contexts = await fetch_agent_authorizations(
+        ...     "https://our-sales-agent.com",
+        ...     ["nytimes.com", "wsj.com", "cnn.com"]
+        ... )
+        >>> for domain, ctx in contexts.items():
+        ...     print(f"{domain}:")
+        ...     print(f"  Property IDs: {ctx.property_ids}")
+        ...     print(f"  Tags: {ctx.property_tags}")
+
+    See Also:
+        ADCPClient.list_authorized_properties: "Push" approach using the agent's API
+
+    Notes:
+        - Silently skips domains where adagents.json is not found or invalid
+        - Only returns domains where the agent is explicitly authorized
+        - For production use with many domains, pass a shared httpx.AsyncClient
+          to enable connection pooling
+    """
+    import asyncio
+
+    # Create tasks to fetch all adagents.json files in parallel
+    async def fetch_authorization_for_domain(
+        domain: str,
+    ) -> tuple[str, AuthorizationContext | None]:
+        """Fetch authorization context for a single domain."""
+        try:
+            adagents_data = await fetch_adagents(domain, timeout=timeout, client=client)
+
+            # Check if agent is authorized
+            if not verify_agent_authorization(adagents_data, agent_url):
+                return (domain, None)
+
+            # Get properties for this agent
+            properties = get_properties_by_agent(adagents_data, agent_url)
+
+            # Create authorization context
+            return (domain, AuthorizationContext(properties))
+
+        except (AdagentsNotFoundError, AdagentsValidationError, AdagentsTimeoutError):
+            # Silently skip domains with missing or invalid adagents.json
+            return (domain, None)
+
+    # Fetch all domains in parallel
+    tasks = [fetch_authorization_for_domain(domain) for domain in publisher_domains]
+    results = await asyncio.gather(*tasks)
+
+    # Build result dictionary, filtering out None values
+    return {domain: ctx for domain, ctx in results if ctx is not None}
diff --git a/tests/test_adagents.py b/tests/test_adagents.py
