socioverse.external_events — real-world event context

The client behind the bundled events capability: structured real-world context (news, indicators, baselines) that the environment stage materializes into a study at build time. The runtime loop never fetches — see External Capabilities for configuration, the fallback chain, and the provenance rules.

materialize_external_events is the entry point most studies use; it writes environment/external_events.json and returns the parsed bundle alongside its path.

socioverse.external_events

ExternalEventsClient — build-time retrieval of external event/macro data for E.

The runtime loop NEVER fetches external data: skills call this at build time (sv-build-environment) and materialize the evidence to studies/<id>/environment/external_events.json; broadcasts/layers are then authored from that file. Priority chain, dropping one tier per failure:

1. "remote" — the production Event MCP (streamable HTTP, stateless JSON mode), configured via SV_EVENT_API_URL (the FULL MCP endpoint, e.g. http://host:9997/event_mcp) + SV_EVENT_API_KEY (bearer). Speaks plain JSON-RPC POSTs (tools/call get_data) — no MCP SDK needed. The server has no query parser (the agent layer was deliberately dropped), so this tier needs explicit months= + sources=; query is provenance only. Returns structured SourceResults only — web evidence is the calling skill's WebSearch job (grounding), never the service's. Ops health lives at <scheme>://<host>/healthz (root, off the MCP path).
2. "local_cache" — an optional local dir of cached per-source payloads (SV_EVENT_LOCAL_CACHE → <dir>/<source>/<YYYY-MM>.json), for offline/dev runs. Plain file reads; requires explicit months + sources since no query parser lives here.
3. "unavailable" — both tiers down: the bundle says so, and the calling skill falls back to Claude WebSearch + hand-authored broadcasts.

Attribution rule (B = f(P, E)): author broadcasts from bundle.results (typed per-source data) or bundle.web_evidence (title/link/date records), never from bundle.summary — that field is another LLM's prose, kept for reference only; injecting it into E would confound behavior attribution.

ExternalEventsBundle

Bases: BaseModel

Materialized external-event evidence for one study environment build.

ExternalEventsClient

ExternalEventsClient(
    api_url: str | None = None,
    api_key: str | None = None,
    local_cache_dir: str | Path | None = None,
    timeout: float = 120.0,
    health_timeout: float = 10.0,
    transport: Callable[..., dict] | None = None,
)

Tiered fetcher: remote Event MCP > local cache dir > unavailable.

Config resolution: explicit ctor args > environment (SV_EVENT_API_URL, SV_EVENT_API_KEY, SV_EVENT_LOCAL_CACHE) > socioverse-beta/.env. The auth key never lives in study artifacts — a ResourceManifest declares the url, the key stays in the environment.

timeout is PER SOURCE-MONTH call: the remote tier fetches source by source (get_source_detail) so one slow upstream (e.g. NYT from a CN-hosted server) can't stall the whole batch — it just becomes an error entry for its key.

Source code in socioverse/external_events.py

def __init__(
    self,
    api_url: str | None = None,
    api_key: str | None = None,
    local_cache_dir: str | Path | None = None,
    timeout: float = 120.0,
    health_timeout: float = 10.0,
    transport: Callable[..., dict] | None = None,
):
    _load_dotenv()
    self.api_url = (api_url or os.environ.get("SV_EVENT_API_URL", "")).rstrip("/")
    self.api_key = api_key or os.environ.get("SV_EVENT_API_KEY", "")
    cache = local_cache_dir or os.environ.get("SV_EVENT_LOCAL_CACHE", "")
    self.local_cache_dir = Path(cache) if cache else None
    self.timeout = timeout
    self.health_timeout = health_timeout
    self._transport = transport or self._http_json

from_manifest classmethod

from_manifest(
    manifest: ResourceManifest,
    name: str = "event_tool",
    **kwargs: Any,
) -> "ExternalEventsClient"

Take the endpoint url from a study's ResourceManifest.mcp_servers entry (transport="http"); the key still comes from the environment.

Source code in socioverse/external_events.py

@classmethod
def from_manifest(
    cls, manifest: ResourceManifest, name: str = "event_tool", **kwargs: Any
) -> "ExternalEventsClient":
    """Take the endpoint url from a study's ResourceManifest.mcp_servers entry
    (transport="http"); the key still comes from the environment."""
    decl = next((s for s in manifest.mcp_servers if s.name == name), None)
    if decl is not None and decl.url:
        kwargs.setdefault("api_url", decl.url)
    return cls(**kwargs)

fetch

fetch(
    query: str,
    months: list[str] | None = None,
    sources: list[str] | None = None,
) -> ExternalEventsBundle

Run the priority chain and return a normalized bundle (never raises).

Source code in socioverse/external_events.py

def fetch(
    self,
    query: str,
    months: list[str] | None = None,
    sources: list[str] | None = None,
) -> ExternalEventsBundle:
    """Run the priority chain and return a normalized bundle (never raises)."""
    months = list(months or [])
    sources = list(sources or [])
    errors: list[dict[str, Any]] = []

    if self.api_url:
        try:
            bundle = self._fetch_remote(query, months, sources)
            bundle.errors = errors + bundle.errors
            return bundle
        except Exception as exc:
            errors.append({"tier": "remote", "endpoint": self.api_url,
                           "error": f"{type(exc).__name__}: {exc}"})
    else:
        errors.append({"tier": "remote", "error": "SV_EVENT_API_URL not configured"})

    if self.local_cache_dir:
        try:
            bundle = self._fetch_local_cache(query, months, sources)
            bundle.errors = errors + bundle.errors
            if bundle.results:
                return bundle
            errors = bundle.errors
        except Exception as exc:
            errors.append({"tier": "local_cache", "dir": str(self.local_cache_dir),
                           "error": f"{type(exc).__name__}: {exc}"})
    else:
        errors.append({"tier": "local_cache", "error": "SV_EVENT_LOCAL_CACHE not configured"})

    return ExternalEventsBundle(
        query=query, requested_months=months, requested_sources=sources,
        provider="unavailable", errors=errors,
    )

month_range

month_range(start: str, end: str) -> list[str]

Inclusive "YYYY-MM" range helper for fetch(months=...).

Source code in socioverse/external_events.py

def month_range(start: str, end: str) -> list[str]:
    """Inclusive "YYYY-MM" range helper for fetch(months=...)."""
    sy, sm = (int(x) for x in start.split("-"))
    ey, em = (int(x) for x in end.split("-"))
    out: list[str] = []
    y, m = sy, sm
    while (y, m) <= (ey, em):
        out.append(f"{y:04d}-{m:02d}")
        m += 1
        if m > 12:
            m, y = 1, y + 1
        if len(out) > 48:
            break
    return out

materialize_external_events

materialize_external_events(
    study_dir: str | Path,
    query: str,
    months: list[str] | None = None,
    sources: list[str] | None = None,
    client: ExternalEventsClient | None = None,
    filename: str = "external_events.json",
) -> tuple[Path, ExternalEventsBundle]

Fetch through the priority chain and persist the evidence bundle into the study's environment/ dir. Returns (path, bundle) so the caller can author broadcasts immediately; check bundle.available to know whether to fall back to Claude WebSearch.

Source code in socioverse/external_events.py

def materialize_external_events(
    study_dir: str | Path,
    query: str,
    months: list[str] | None = None,
    sources: list[str] | None = None,
    client: ExternalEventsClient | None = None,
    filename: str = "external_events.json",
) -> tuple[Path, ExternalEventsBundle]:
    """Fetch through the priority chain and persist the evidence bundle into the
    study's environment/ dir. Returns (path, bundle) so the caller can author
    broadcasts immediately; check `bundle.available` to know whether to fall back
    to Claude WebSearch."""
    client = client or ExternalEventsClient()
    bundle = client.fetch(query, months=months, sources=sources)
    path = Path(study_dir) / "environment" / filename
    write_artifact(path, bundle)
    return path, bundle