avoigt1121
Claude Opus 5.5
docs: name no institution in the public repo (the client's Tech Transfer rule)
081f1c7 | """The two config files every other module reads: the tool allow-list and the source list. | |
| `config/tools_allowlist.yaml` is the approved surface (ADR-0001 §7): which ToolUniverse tools may | |
| ever load, and which of this agent's endpoints uses each. `config/sources.yaml` is the honest egress statement | |
| (ADR-0001 §6): the hosts each source's tools reach, the licence stamped on its rows, and its cache | |
| and rate settings. Both are validated here once, so a malformed file stops the Space at import | |
| rather than surfacing as a half-stamped row later. | |
| """ | |
| from __future__ import annotations | |
| import functools | |
| from pathlib import Path | |
| from typing import Any | |
| import yaml | |
| ROOT = Path(__file__).resolve().parent.parent | |
| CONFIG_DIR = ROOT / "config" | |
| # Frozen by ADR-0001 §4 (accepted 2026-10-20). s2 builds them; the names are fixed now so the | |
| # loaded tool set can be checked before any endpoint code exists. | |
| ENDPOINTS = ( | |
| "drugs_for_targets", | |
| "trials_for_genes", | |
| "annotate_genes", | |
| "pathways_for_genes", | |
| "targets_for_disease", | |
| ) | |
| _TOOL_FIELDS = ("name", "source", "endpoints", "review") | |
| _SOURCE_FIELDS = { | |
| "name": str, | |
| "hosts": list, | |
| "licence": str, | |
| "licence_url": str, | |
| "licence_checked": str, | |
| "ttl_seconds": int, | |
| "rate_per_second": (int, float), | |
| "burst": int, | |
| "timeout_seconds": (int, float), | |
| "coverage_notes": list, | |
| "citation": str, | |
| } | |
| class ConfigError(ValueError): | |
| """A config file is malformed. Raised at import, never at request time.""" | |
| def _read(name: str) -> dict[str, Any]: | |
| with open(CONFIG_DIR / name, encoding="utf-8") as fh: | |
| data = yaml.safe_load(fh) | |
| if not isinstance(data, dict): | |
| raise ConfigError(f"{name}: expected a mapping at the top level") | |
| return data | |
| def sources() -> dict[str, dict[str, Any]]: | |
| """Every declared source, keyed by its short id (`opentargets`, `chembl`, …).""" | |
| data = _read("sources.yaml").get("sources") | |
| if not isinstance(data, dict) or not data: | |
| raise ConfigError("sources.yaml: `sources` must be a non-empty mapping") | |
| for sid, entry in data.items(): | |
| for field, kind in _SOURCE_FIELDS.items(): | |
| if not isinstance(entry.get(field), kind): | |
| raise ConfigError(f"sources.yaml: {sid}.{field} missing or not {kind}") | |
| if not entry["hosts"] or not all(isinstance(h, str) and h for h in entry["hosts"]): | |
| raise ConfigError(f"sources.yaml: {sid}.hosts must list at least one host") | |
| if not entry["licence"].strip() or not entry["licence_url"].startswith("https://"): | |
| raise ConfigError(f"sources.yaml: {sid} needs a licence and an https licence_url") | |
| if entry["rate_per_second"] <= 0 or entry["burst"] < 1 or entry["ttl_seconds"] <= 0: | |
| raise ConfigError(f"sources.yaml: {sid} has a non-positive rate, burst or TTL") | |
| return data | |
| def source(source_id: str) -> dict[str, Any]: | |
| try: | |
| return sources()[source_id] | |
| except KeyError: | |
| raise ConfigError(f"unknown source {source_id!r}") from None | |
| def allowlist() -> dict[str, Any]: | |
| """The parsed allow-list: `tooluniverse_version` plus `tools` keyed by tool name.""" | |
| data = _read("tools_allowlist.yaml") | |
| version = data.get("tooluniverse_version") | |
| if not isinstance(version, str) or not version: | |
| raise ConfigError("tools_allowlist.yaml: `tooluniverse_version` must be a string") | |
| tools: dict[str, dict[str, Any]] = {} | |
| for entry in data.get("tools") or []: | |
| if not all(field in entry for field in _TOOL_FIELDS): | |
| raise ConfigError(f"tools_allowlist.yaml: entry missing {_TOOL_FIELDS}: {entry!r}") | |
| name = entry["name"] | |
| if name in tools: | |
| raise ConfigError(f"tools_allowlist.yaml: {name} is listed twice") | |
| if entry["source"] not in sources(): | |
| raise ConfigError( | |
| f"tools_allowlist.yaml: {name} names undeclared source {entry['source']!r}" | |
| ) | |
| unknown = set(entry["endpoints"]) - set(ENDPOINTS) | |
| if unknown: | |
| raise ConfigError( | |
| f"tools_allowlist.yaml: {name} names unknown endpoints {sorted(unknown)}" | |
| ) | |
| if not str(entry["review"]).strip(): | |
| raise ConfigError(f"tools_allowlist.yaml: {name} has no review note") | |
| tools[name] = entry | |
| if not tools: | |
| raise ConfigError("tools_allowlist.yaml: no tools") | |
| return {"tooluniverse_version": version, "tools": tools} | |
| def pinned_tooluniverse_version() -> str: | |
| return allowlist()["tooluniverse_version"] | |
| def allowed_tool_names() -> frozenset[str]: | |
| """Every approved tool (the memo's 28), loaded or not.""" | |
| return frozenset(allowlist()["tools"]) | |
| def endpoint_tool_names() -> frozenset[str]: | |
| """The tools at least one endpoint uses: exactly the set that is loaded (ADR-0001 §7).""" | |
| return frozenset(name for name, t in allowlist()["tools"].items() if t["endpoints"]) | |
| def tools_for_endpoint(endpoint: str) -> tuple[str, ...]: | |
| if endpoint not in ENDPOINTS: | |
| raise ConfigError(f"unknown endpoint {endpoint!r}") | |
| return tuple( | |
| sorted(name for name, t in allowlist()["tools"].items() if endpoint in t["endpoints"]) | |
| ) | |
| def source_of(tool: str) -> str: | |
| try: | |
| return allowlist()["tools"][tool]["source"] | |
| except KeyError: | |
| raise ConfigError(f"{tool!r} is not on the allow-list") from None | |