atlas2-mcp/app/resolve.py

from __future__ import annotations

import os
import time
import uuid
from dataclasses import dataclass
from typing import Any, Awaitable, Callable


CallToolFn = Callable[[str, dict[str, Any]], Awaitable[dict[str, Any]]]


def _extract_bindings(result_payload: Any) -> list[dict[str, Any]]:
    """Best-effort extraction for Virtuoso/MCP-style SPARQL results."""
    if isinstance(result_payload, dict):
        return result_payload.get("results", {}).get("bindings", []) or []
    return []


def _to_float(value: Any) -> float:
    try:
        return float(value)
    except Exception:
        return 0.0


def _now_iso() -> str:
    # Avoid datetime imports; keep it lightweight.
    import datetime

    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def _build_candidates_query(*, subject: str, max_candidates: int, graph_iri: str) -> str:
    # Scaffolding query: adjust the predicate/shape once the remote schema is fixed.
    # Keep it deterministic and parameterized by the provided subject string.
    safe = subject.replace("\\", "\\\\").replace('"', '\\"')
    return f"""
PREFIX atlas: <http://world.eu.org/atlas_ontology#>
SELECT ?id ?label ?type ?source ?confidence ?description ?uri
WHERE {{
  GRAPH <{graph_iri}> {{
    ?entity a atlas:Entity ;
            atlas:canonicalLabel ?label .

    FILTER(LCASE(STR(?label)) = LCASE(\"{safe}\"))

    OPTIONAL {{ ?entity atlas:hasCanonicalType ?type . }}
    OPTIONAL {{ ?entity atlas:canonicalDescription ?description . }}

    BIND(STR(?entity) AS ?uri)
    BIND(STRAFTER(STR(?entity), '#') AS ?id)
    BIND(0.9 AS ?confidence)
    BIND("sparql" AS ?source)
  }}
}}
LIMIT {max_candidates}
""".strip()


@dataclass
class ResolveService:
    call_tool: CallToolFn | None = None

    async def _call_tool(self, tool_name: str, payload: dict[str, Any]) -> dict[str, Any]:
        if self.call_tool is None:
            # Important: default behavior is a stub. This scaffolding should run
            # safely without requiring a live remote MCP/Sparql backend.
            raise RuntimeError(
                "REMOTE_SPASQL_MCP_NOT_CONFIGURED (stub). "
                "Inject call_tool in tests or wire a real RemoteSparqlClient explicitly."
            )
        return await self.call_tool(tool_name, payload)

    async def resolve(
        self,
        *,
        subject: str,
        context: dict[str, Any] | None,
        constraints: dict[str, Any] | None,
        hints: dict[str, Any] | None,
        debug: dict[str, Any] | None,
    ) -> dict[str, Any]:
        # Stubbed implementation for “works first, decide logic later”.
        # We intentionally ignore inputs until you confirm the app structure.
        return {"status": "ok"}