tooluniverse.wikidata_sparql_tool 源代码

import requests
from .base_tool import BaseTool
from .tool_registry import register_tool




@register_tool("WikidataSPARQLTool")
class WikidataSPARQLTool(BaseTool):
    """
    Run SPARQL queries against Wikidata (powering Scholia views).

    Parameters (arguments):
        sparql (str): SPARQL query string
        max_results (int): Optional result limit override
    """



    def __init__(self, tool_config):
        super().__init__(tool_config)
        self.endpoint = "https://query.wikidata.org/sparql"




    def run(self, arguments=None):
        arguments = arguments or {}
        sparql = arguments.get("sparql")
        max_results = arguments.get("max_results")
        if not sparql:
            return {"status": "error", "error": "`sparql` parameter is required."}
        if max_results:
            # naive limit appending if not present
            if "limit" not in sparql.lower():
                sparql = f"{sparql}\nLIMIT {int(max_results)}"

        headers = {
            "Accept": "application/sparql-results+json",
            "User-Agent": "ToolUniverse/1.0 (https://github.com)",
        }
        try:
            resp = requests.get(
                self.endpoint,
                params={"query": sparql, "format": "json"},
                headers=headers,
                timeout=30,
            )
            resp.raise_for_status()
            data = resp.json()
        except requests.RequestException as e:
            return {
                "status": "error",
                "error": "Network/API error calling Wikidata SPARQL",
                "reason": str(e),
            }
        except ValueError:
            ct = resp.headers.get("content-type", "")
            return {
                "status": "error",
                "error": "Failed to decode SPARQL response as JSON",
                "content_type": ct,
                "response_snippet": resp.text[:200],
                "retryable": "text/html" in ct or resp.text.lstrip().startswith("<"),
                "suggestion": "Wikidata SPARQL endpoint may be under maintenance. Retry in a few minutes.",
            }

        bindings = data.get("results", {}).get("bindings", [])
        # Normalize by unwrapping "value" fields
        normalized = []
        for b in bindings:
            row = {}
            for k, v in b.items():
                row[k] = v.get("value")
            normalized.append(row)
        return normalized