148 lines
5.3 KiB
Python
148 lines
5.3 KiB
Python
"""Client for the `market:catalog` semantic index (memory-api).
|
|
|
|
Query contract: specs/001-market-first-wedge/contracts/catalog-index.md
|
|
|
|
Adapted 2026-07-02 to the REAL memory-api surface (agent.memory.system):
|
|
`POST /memory/search` with `{query, agent_id, limit}` — the partition is the
|
|
`agent_id` namespace, and each result carries the listing in `metadata`.
|
|
|
|
DEVIATION from catalog-index.md (recorded): memory-api's server-side filter is a
|
|
scalar `tenant_id`, but listings carry a `tenant_visibility` ARRAY. Until the
|
|
memory-api grows array filters (fedlearn T13 follow-up), the visibility filter
|
|
`tenant_visibility ⊇ {caller tenant | "*"}` is enforced HERE, in the one shared
|
|
reader client that every surface (CLI, Store, Scout) goes through — never in
|
|
per-surface code.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import httpx
|
|
|
|
# Overridable so a full catalog rebuild can land in a fresh partition and readers
|
|
# switch by config (constitution II rebuildability; see docs/runbook.md).
|
|
PARTITION = os.environ.get("M2_MARKET_CATALOG_PARTITION", "market:catalog")
|
|
|
|
|
|
class CatalogClient:
|
|
"""Tenant-scoped semantic search over the market:catalog partition."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
api_key: str,
|
|
tenant: str,
|
|
*,
|
|
search_path: str = "/memory/search",
|
|
timeout: float = 30.0,
|
|
transport: httpx.BaseTransport | None = None,
|
|
) -> None:
|
|
self.tenant = tenant
|
|
self.search_path = search_path
|
|
self._client = httpx.Client(
|
|
base_url=base_url,
|
|
headers={"X-API-Key": api_key},
|
|
timeout=timeout,
|
|
transport=transport,
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
def __enter__(self) -> "CatalogClient":
|
|
return self
|
|
|
|
def __exit__(self, *exc_info: object) -> None:
|
|
self.close()
|
|
|
|
def _visible(self, listing: dict) -> bool:
|
|
visibility = listing.get("tenant_visibility") or []
|
|
return "*" in visibility or self.tenant in visibility
|
|
|
|
def search(self, query_text: str, limit: int = 10) -> list[dict]:
|
|
"""Semantic query against market:catalog, visibility-filtered to `self.tenant`.
|
|
|
|
Returns a list of `{listing_id, score, listing}` ordered by hybrid score.
|
|
"""
|
|
response = self._client.post(
|
|
self.search_path,
|
|
json={
|
|
"query": query_text,
|
|
"agent_id": PARTITION,
|
|
"memory_types": ["semantic"],
|
|
# Overfetch so the visibility filter doesn't starve the result set.
|
|
"limit": max(limit * 3, limit),
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
items = []
|
|
for result in response.json().get("results", []):
|
|
meta = result.get("metadata") or {}
|
|
listing = meta.get("listing")
|
|
listing_id = meta.get("listing_id")
|
|
if not listing or not listing_id:
|
|
continue
|
|
if not self._visible(listing):
|
|
continue
|
|
items.append(
|
|
{"listing_id": listing_id, "score": result.get("score"), "listing": listing}
|
|
)
|
|
if len(items) >= limit:
|
|
break
|
|
return items
|
|
|
|
def get(self, listing_id: str) -> dict | None:
|
|
"""Fetch one listing's payload by id, or None if not found/visible."""
|
|
for item in self.search(listing_id, limit=10):
|
|
if item.get("listing_id") == listing_id:
|
|
return item.get("listing")
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Inline verification against the real response shape.
|
|
def _handler(request: httpx.Request) -> httpx.Response:
|
|
import json as _json
|
|
|
|
body = _json.loads(request.content)
|
|
assert body["agent_id"] == PARTITION, body
|
|
assert body["query"], body
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"results": [
|
|
{
|
|
"id": "mem-1",
|
|
"content": "PDF Export ...",
|
|
"score": 0.87,
|
|
"metadata": {
|
|
"listing_id": "lst_pdf-export",
|
|
"listing": {"name": "PDF Export", "tenant_visibility": ["*"]},
|
|
},
|
|
},
|
|
{
|
|
"id": "mem-2",
|
|
"content": "tenant-only thing",
|
|
"score": 0.5,
|
|
"metadata": {
|
|
"listing_id": "lst_private",
|
|
"listing": {"name": "Private", "tenant_visibility": ["other-tenant"]},
|
|
},
|
|
},
|
|
],
|
|
"routing": {},
|
|
},
|
|
)
|
|
|
|
client = CatalogClient(
|
|
base_url="https://memory.machinemachine.ai",
|
|
api_key="mem_test_key",
|
|
tenant="m2-core",
|
|
transport=httpx.MockTransport(_handler),
|
|
)
|
|
results = client.search("branded pdf report", limit=5)
|
|
assert [r["listing_id"] for r in results] == ["lst_pdf-export"], results # filtered
|
|
assert client.get("lst_pdf-export") == {"name": "PDF Export", "tenant_visibility": ["*"]}
|
|
assert client.get("lst_private") is None # invisible to this tenant
|
|
print("OK:", results)
|