"""The specialists this agent may consult, and how to reach one. A specialist is described here by what it *decides*, not by what data it holds. That is the difference the estate is built on: an A2A call returns an answer, so the orchestrator learns "38 products, the dearest is S10_1949" and never gains read access to the catalog domain. Adding a specialist is three things: an entry below, an audience scope granted to `classic-models-agent-assistant` in the realm, and nothing else. If a specialist appears here without the realm grant, every call to it fails at the token exchange with `Invalid scopes` — loudly, and before any data moves. """ from __future__ import annotations import os from dataclasses import dataclass, replace @dataclass(frozen=True) class Specialist: """One agent this orchestrator can delegate to.""" name: str url: str audience_scope: str serves: str """What it is for, in the words the model routing on it will read.""" decides: tuple[str, ...] """The questions it can settle. Deliberately narrow. These become the tool description. A specialist advertised as "ask it anything about products" turns the orchestrator into a data proxy by the back door — the very thing the topology exists to prevent. """ CATALOGUE = Specialist( name="catalogue", url="http://catalogue-agent.agents.svc.cluster.local:8080/", audience_scope="classic-models-agent-catalogue-audience", serves="product management: products, product lines, pricing, discontinuation", decides=( "how many products a product line holds, and which are the dearest or cheapest", "a product's details: name, vendor, scale, stock, buy price, MSRP", "whether a product sells well enough to justify a pricing or discontinuation decision", "creating a product, which it holds for the user's approval before writing", ), ) #: Everything the design foresees, minus what is not built. Ventes, Support, #: Approvisionnement and Recouvrement belong here as they arrive; the graph #: needs no change for them, only this list and a realm grant. SPECIALISTS: tuple[Specialist, ...] = (CATALOGUE,) def by_name(name: str) -> Specialist | None: """The specialist called `name`, with its URL overridable per environment. `ASSISTANT_URL_` exists for running the orchestrator outside the cluster against a port-forwarded specialist. In the cluster nothing sets it and the Service DNS name below is used as written. """ found = next((s for s in SPECIALISTS if s.name == name), None) if found is None: return None override = os.environ.get(f"ASSISTANT_URL_{found.name.upper()}") return replace(found, url=override) if override else found def describe(specialist: Specialist) -> str: """The tool description the routing model reads.""" lines = [f"Consult the {specialist.name} agent. It serves {specialist.serves}."] lines.append("It can settle:") lines.extend(f"- {d}" for d in specialist.decides) lines.append( "Ask it a complete question in plain language, as you would ask a colleague. " "It answers; it does not hand back raw records." ) return "\n".join(lines)