feat(directory): support source-aware directory discovery

Author: nkavian

README.md

@@ -13,7 +13,7 @@ Services and navigating their Offerings.
 ODP separates two levels of discovery:
 
 1. An Agent searches the canonical Directory for Services.
-2. The Agent inspects a Service's live ODP document and navigates that Service's Collections and
+2. For a native ODP source, the Agent inspects the Service's live ODP document and navigates its Collections and
    Offerings.
 
 The Directory does not copy every Service catalog. Catalog searches go directly to each Service.
@@ -77,17 +77,36 @@ asyncio.run(main())
 an explicit list must be nonempty and distinct. Filters apply to the owning Service's metadata.
 Omit the query to browse. The default and maximum result limit are 100.
 
-A Collection is identified by its owning Service origin and case-sensitive Collection ID.
-Inspect that Service's live ODP document, then call `ServiceClient.get_collection()` for current
-details. The result's `indexed_at` describes Collection freshness; `service.indexed_at` describes
-parent freshness. `service.service_id` is the Directory's Service identifier. A Service result can
+A Collection is identified by its owning `service.service_id` and case-sensitive Collection ID.
+Different OpenAPI documents can share an API origin without being the same Directory Service.
+When `service.source.type == "odp"`, inspect that Service's live ODP document, then call
+`ServiceClient.get_collection()` for current details. OpenAPI Collections are Directory presentation
+groups, not ODP operation targets. The result's `indexed_at` describes Collection freshness;
+`service.indexed_at` describes parent freshness. `service.service_id` is the Directory's Service identifier. A Service result can
 have `available_through` platform attribution; a Collection's attribution is its owning `service`.
 
 Malformed known results are omitted and reported in `issues` with their original response index.
 Unknown future types retain their full JSON in `UnknownResult.raw`; do not treat them as Services
 or execute their metadata. Additional fields are available through `additional`. Directory
 metadata does not replace inspection of the Service's own document.
 
+Mixed results use `DirectoryIndexedService`, separate from the native `DirectoryService` returned
+by `search_services()`. Each mixed Service requires `service_id`, `service_origin`, `name`,
+`indexed_at` and `source`. Imported descriptions and languages are optional; missing lists become
+empty lists. Imported results do not expose native ODP operations. Native results retain ODP
+validation. Unverified execution fields such as `http` and `payment_origins` are not returned.
+
+`DirectorySource` identifies the document used for discovery:
+
+- `type` is `"odp"`, `"openapi"`, or an unknown future string. Unknown formats remain readable
+  but must not be passed to ODP operations.
+- `url` is the exact document URL, including path and query. It can differ from the API origin;
+  do not reconstruct it from `service_origin`.
+- `x402_discovery` records supporting fixed-path x402 discovery, not proof that an endpoint
+  accepts payments. Advertised protocol evidence remains in `protocols`.
+
+The client does not fetch or execute OpenAPI documents.
+
 Mixed search does not currently offer continuation. Missing `next` does **not** mean every match
 was returned. Refine the query or filters when needed. `continue_search(next)` follows one opaque
 same-origin reference if the server supplies one; the SDK does not invent continuations. Facets
@@ -105,7 +124,32 @@ are candidate search queries, not resource identifiers.
 
 See the [runnable canonical Directory example](examples/README.md#canonical-directory-discovery).
 
-## Search only Services
+### Filter by source
+
+```python
+from offering_protocol.directory import (
+    DirectoryClient,
+    ResourceSearchRequest,
+    ServiceFilters,
+    SuggestionRequest,
+)
+
+
+async def discover_openapi() -> None:
+    filters = ServiceFilters(sources=["openapi"])
+    async with DirectoryClient() as directory:
+        results = await directory.search(ResourceSearchRequest(query="weather", filters=filters))
+        names = await directory.suggest(SuggestionRequest(prefix="we", filters=filters))
+        print(results.items, names)
+```
+
+Omitting `sources` includes all formats. An explicit list must contain one or both distinct
+`"odp"` and `"openapi"` values. Sources are alternatives, combined with other filter categories
+using AND. Collections inherit their owning Service's source. Unsupported source filter values
+are rejected. Native `search_services()` accepts the filter but remains ODP-only: an OpenAPI-only
+filter returns no native matches.
+
+## Search only native ODP Services
 
 `DirectoryClient` uses the one canonical production Directory. Pass `Environment.SANDBOX` when
 working against InFlow's sandbox; the endpoint itself is not configurable.
@@ -143,6 +187,9 @@ and trust protocols. Use `suggest_services()` to discover Service-only keyword c
 
 ### API migration
 
+- Mixed results use `DirectoryIndexedService` with required source metadata. Missing sources are
+  reported as item issues, not assumed to be ODP. Native Service-only models are unchanged.
+- Imported `description` and `language` can be `None`; check the source before ODP navigation.
 - Service-only `search()` calls become `search_services()`, and `continue_search()` calls become
   `continue_search_services()`.
 - Aggregating `search_services(request, options)` calls become `collect_services(request, options)`.
@@ -385,8 +432,9 @@ Protocol models preserve additive members in `model.additional` and round-trip t
 `model.to_dict()`. Parsing remains strict for normative constraints and fields that prohibit unknown
 members.
 
-Directory records retain additional metadata such as branding and MCP endpoints. These are discovery
-hints, not authorization or authoritative routing data. The default Agent factory uses the record's
+Native Service-only records retain additional metadata such as branding and MCP endpoints. Mixed
+results omit unverified execution fields. Directory metadata is not authorization or authoritative
+routing data. The default Agent factory uses the native record's
 `service_origin` and retrieves that Service's own document before making catalog requests.
 
 Handle the narrowest error that the application can act upon and use the role's base error for the

examples/README.md

@@ -39,6 +39,7 @@ uv run python examples/directory.py sandbox weather
 Use `production` for the production Directory. Omit `weather` to browse. This example requires a
 deployment with `/v1/directory/search`. It requests at most five mixed results, displays Service
 and Collection names, reports unusable or unknown results, and retrieves Collection details only
-after inspecting the owning Service's advertised anonymous support. It does not enroll, pay or
+for ODP sources after inspecting the owning Service's advertised anonymous support. Imported
+Collections print their exact document URL without ODP calls. It does not enroll, pay or
 invoke Actions. Unlike the local Service example above, this uses the real Directory.
 The server's bounded result list does not promise every matching result is included.

examples/directory.py

@@ -25,11 +25,15 @@ async def discover(environment: Environment, query: str) -> None:
     for item in response.items:
         if isinstance(item, ServiceResult):
             print(f"Service: {item.service.name} ({item.service.service_origin})")
+            print(f"Discovery document: {item.service.source.url}")
         elif isinstance(item, CollectionResult):
             print(
                 f"Collection: {item.collection.name} "
                 f"({item.collection.id}, through {item.service.service_origin})"
             )
+            if item.service.source.type != "odp":
+                print(f"Discovery document: {item.service.source.url}")
+                continue
             async with ServiceClient(item.service.service_origin) as service:
                 inspection = await service.inspect()
                 if any(

scripts/verify-consumer.sh

@@ -83,6 +83,14 @@ assert directory.__name__ == "offering_protocol.directory"
 assert service.__name__ == "offering_protocol.service"
 request = directory.ResourceSearchRequest(types=["collection"])
 assert request.to_dict() == {"types": ["collection"]}
+filters = directory.ServiceFilters(sources=["openapi"])
+assert filters.to_dict() == {"sources": ["openapi"]}
+source = directory.DirectorySource(type="openapi", url="https://example.com/api.json", x402_discovery=False)
+record = directory.DirectoryIndexedService(
+    indexed_at="2026-09-23T12:00:00Z", name="Example", service_id="example",
+    service_origin="https://example.com", source=source,
+)
+assert record.source.url == source.url and record.operations == []
 document = core.parse_service_document(
     b'{"description":"Consumer smoke test","http":{"endpoint_base":"/odp"},'
     b'"language":"en","localizations":["en"],"name":"Consumer",'

src/offering_protocol/directory/__init__.py

@@ -8,9 +8,11 @@
 from offering_protocol.directory.models import (
     CollectionResult,
     CollectionSummary,
+    DirectoryIndexedService,
     DirectoryIssue,
     DirectoryResult,
     DirectoryService,
+    DirectorySource,
     Environment,
     Facet,
     Facets,
@@ -42,10 +44,12 @@
     "CollectionSummary",
     "DirectoryClient",
     "DirectoryError",
+    "DirectoryIndexedService",
     "DirectoryIssue",
     "DirectoryRequestError",
     "DirectoryResult",
     "DirectoryService",
+    "DirectorySource",
     "Environment",
     "Facet",
     "Facets",

src/offering_protocol/directory/client.py

@@ -332,6 +332,10 @@ def _validate_search_request(request: SearchRequest) -> None:
         raise DirectoryError(
             "query must contain at most 512 characters without surrounding whitespace"
         )
+    if request.filters is not None and request.filters.sources is not None:
+        sources = request.filters.sources
+        if not 1 <= len(sources) <= 2 or len(set(sources)) != len(sources):
+            raise DirectoryError("sources must contain one or two distinct odp or openapi values")
     if request.filters is not None and (
         len(request.filters.keywords) > 32
         or any(not keyword or len(keyword) > 64 for keyword in request.filters.keywords)

src/offering_protocol/directory/models.py

@@ -50,6 +50,7 @@ class ServiceFilters(OdpModel):
     keywords: list[str] = Field(default_factory=list)
     operations: list[OperationFilter] = Field(default_factory=list)
     payments: list[PaymentFilter] = Field(default_factory=list)
+    sources: list[Literal["odp", "openapi"]] | None = None
     trust: list[TrustProtocol] = Field(default_factory=list)
 
 
@@ -84,6 +85,30 @@ def service_id(self) -> str | None:
         return value if isinstance(value, str) else None
 
 
+class DirectorySource(OdpModel):
+    type: str
+    url: str
+    x402_discovery: bool
+
+
+class DirectoryIndexedService(OdpModel):
+    description: str | None = None
+    documentation_url: str | None = None
+    indexed_at: str
+    keywords: list[str] = Field(default_factory=list)
+    language: str | None = None
+    localizations: list[str] = Field(default_factory=list)
+    name: str
+    operations: list[OperationDescriptor] = Field(default_factory=list)
+    protocols: ServiceProtocols | None = None
+    service_id: str
+    service_origin: str
+    source: DirectorySource
+    status_url: str | None = None
+    support_url: str | None = None
+    website_url: str | None = None
+
+
 class ServiceReference(OdpModel):
     service_id: str
     service_origin: str
@@ -98,14 +123,14 @@ class CollectionSummary(OdpModel):
 
 class ServiceResult(OdpModel):
     type: Literal["service"]
-    service: DirectoryService
+    service: DirectoryIndexedService
     indexed_at: str
     available_through: ServiceReference | None = None
 
 
 class CollectionResult(OdpModel):
     type: Literal["collection"]
-    service: DirectoryService
+    service: DirectoryIndexedService
     indexed_at: str
     collection: CollectionSummary
 

src/offering_protocol/directory/results.py

@@ -20,6 +20,7 @@
     ServiceResult,
     UnknownResult,
 )
+from offering_protocol.directory.sources import read_source, validate_imported_service
 
 _OBJECT = TypeAdapter(dict[str, JsonValue])
 
@@ -56,7 +57,7 @@ def _result(value: JsonValue) -> DirectoryResult:
     _text(service, "service_id", 128)
     _origin(service)
     _timestamp(service)
-    candidate = dict(service)
+    source = read_source(service.get("source"))
     for name in (
         "branding",
         "http",
@@ -65,7 +66,18 @@ def _result(value: JsonValue) -> DirectoryResult:
         "payment_origins",
         "search_capabilities",
     ):
-        candidate.pop(name, None)
+        service.pop(name, None)
+    if source.type == "odp":
+        _native_service(service)
+    else:
+        validate_imported_service(service)
+    raw["service"] = service
+    return _finish_result(raw, kind)
+
+
+def _native_service(service: dict[str, JsonValue]) -> None:
+    candidate = dict(service)
+    candidate.pop("source")
     document = parse_agent_service_document(
         json.dumps({**candidate, "odp_version": "1.0", "http": {"endpoint_base": "/"}})
     )
@@ -74,7 +86,9 @@ def _result(value: JsonValue) -> DirectoryResult:
         service.pop("protocols", None)
     else:
         service["protocols"] = document.protocols.to_dict()
-    raw["service"] = service
+
+
+def _finish_result(raw: dict[str, JsonValue], kind: str) -> DirectoryResult:
     if kind == "service":
         if "available_through" in raw:
             reference = _OBJECT.validate_python(raw["available_through"])

src/offering_protocol/directory/sources.py

@@ -0,0 +1,91 @@
+"""Source-aware metadata validation for mixed Directory results."""
+
+from __future__ import annotations
+
+from ipaddress import ip_address
+from urllib.parse import urlsplit
+
+from pydantic import JsonValue, TypeAdapter
+
+from offering_protocol.core import derive_service_origin, validate_value
+from offering_protocol.directory.addresses import is_public
+from offering_protocol.directory.models import DirectorySource
+
+_OBJECT = TypeAdapter(dict[str, JsonValue])
+
+
+def read_source(value: JsonValue) -> DirectorySource:
+    source = DirectorySource.model_validate(value, strict=True)
+    if not source.type.strip() or len(source.type) > 128:
+        raise ValueError("source.type must be a nonempty string of at most 128 characters")
+    url = source.url
+    if (
+        len(url) > 2048
+        or not url.lower().startswith("https://")
+        or any(character.isspace() for character in url)
+        or "#" in url
+    ):
+        raise ValueError("source.url must be an HTTPS document URL without a fragment")
+    origin = derive_service_origin(url)
+    host = urlsplit(origin).hostname or ""
+    if host.rstrip(".") == "localhost" or host.rstrip(".").endswith(".localhost"):
+        raise ValueError("source.url must have a public host")
+    try:
+        address = ip_address(host)
+    except ValueError:
+        return source
+    if not is_public(address):
+        raise ValueError("source.url must have a public host")
+    return source
+
+
+def validate_imported_service(service: dict[str, JsonValue]) -> None:
+    name = service.get("name")
+    if not isinstance(name, str) or not name.strip() or len(name) > 128:
+        raise ValueError(
+            "Imported Service name must be a nonempty string of at most 128 characters"
+        )
+    for field in (
+        "description",
+        "documentation_url",
+        "language",
+        "status_url",
+        "support_url",
+        "website_url",
+    ):
+        if field in service and not isinstance(service[field], str):
+            raise ValueError(f"{field} must be a string")
+    service.pop("operations", None)
+    if "protocols" not in service:
+        return
+    protocols = _OBJECT.validate_python(service["protocols"])
+    retained: dict[str, JsonValue] = {}
+    for category, known, schema in (
+        ("enrollment", {"aep"}, "enrollment-protocol.schema.json"),
+        ("payments", {"mpp", "x402"}, "payment-protocol.schema.json"),
+        ("trust", {"tap"}, "trust-protocol.schema.json"),
+    ):
+        if category not in protocols:
+            continue
+        values = protocols[category]
+        if not isinstance(values, list) or not values:
+            raise ValueError(f"protocols.{category} must be a nonempty array")
+        selected: list[JsonValue] = []
+        names: set[str] = set()
+        for value in values:
+            descriptor = _OBJECT.validate_python(value)
+            name = descriptor.get("name")
+            if not isinstance(name, str) or not name.strip() or len(name) > 128:
+                raise ValueError(
+                    "Protocol name must be a nonempty string of at most 128 characters"
+                )
+            if name not in known:
+                continue
+            if name in names:
+                raise ValueError(f"Duplicate {category} descriptor")
+            names.add(name)
+            validate_value(descriptor, schema, category)
+            selected.append(descriptor)
+        if selected:
+            retained[category] = selected
+    service["protocols"] = retained