sparql_query
Execute SPARQL queries on Wikidata to retrieve structured data in JSON, CSV, or XML formats. Specify query, response format, and result limits for precise data extraction.
Instructions
Execute a SPARQL query against Wikidata
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| format | No | Response format | json |
| limit | No | Maximum number of results (default: 100, max: 1000) | |
| query | Yes | SPARQL query |
Input Schema (JSON Schema)
{
"properties": {
"format": {
"default": "json",
"description": "Response format",
"enum": [
"json",
"csv",
"xml"
],
"type": "string"
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100, max: 1000)",
"maximum": 1000,
"type": "integer"
},
"query": {
"description": "SPARQL query",
"type": "string"
}
},
"required": [
"query"
],
"type": "object"
}
Implementation Reference
- mcp_wikidata/wikidata_client.py:145-192 (handler)Core handler function that executes the SPARQL query logic: appends LIMIT if needed, sends HTTP request to Wikidata SPARQL endpoint with proper headers, parses JSON or returns text, and handles timeouts, rate limits, server errors, and connectivity issues.async def sparql_query( self, query: str, format: str = "json", limit: int = 100 ) -> Dict[str, Any]: if "LIMIT" not in query.upper(): query += f" LIMIT {min(limit, 1000)}" params = { "query": query, "format": format } try: response = await self.session.get( self.config.sparql_endpoint, params=params, headers={ "Accept": f"application/sparql-results+{format}", "User-Agent": self.config.user_agent } ) response.raise_for_status() if format == "json": return response.json() else: return {"result": response.text} except asyncio.TimeoutError: raise Exception(f"SPARQL query timed out after {self.config.timeout} seconds. Try simplifying your query or increasing the timeout.") except httpx.TimeoutException: raise Exception(f"SPARQL query timed out after {self.config.timeout} seconds. Try simplifying your query or increasing the timeout.") except httpx.HTTPStatusError as e: if e.response.status_code == 500: raise Exception(f"SPARQL server error (500). Your query may have syntax errors or be too complex: {query[:100]}...") elif e.response.status_code == 429: raise Exception("SPARQL service is rate limiting requests. Please wait and try again.") else: raise Exception(f"SPARQL HTTP error {e.response.status_code}: {e.response.text[:200]}") except httpx.ConnectError: raise Exception(f"Cannot connect to SPARQL endpoint {self.config.sparql_endpoint}. Check your internet connection.") except Exception as e: if "timeout" in str(e).lower(): raise Exception(f"SPARQL query timed out after {self.config.timeout} seconds. Try simplifying your query or increasing the timeout.") raise Exception(f"SPARQL query failed: {str(e)}")
- mcp_wikidata/tools.py:77-102 (registration)MCP tool registration for 'sparql_query', including name, description, and full input schema definition.Tool( name="sparql_query", description="Execute a SPARQL query against Wikidata", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SPARQL query" }, "format": { "type": "string", "description": "Response format", "enum": ["json", "csv", "xml"], "default": "json" }, "limit": { "type": "integer", "description": "Maximum number of results (default: 100, max: 1000)", "default": 100, "maximum": 1000 } }, "required": ["query"] } ),
- mcp_wikidata/tools.py:80-100 (schema)Input schema defining parameters for the sparql_query tool: required 'query' string, optional 'format' (json/csv/xml), and 'limit' integer.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SPARQL query" }, "format": { "type": "string", "description": "Response format", "enum": ["json", "csv", "xml"], "default": "json" }, "limit": { "type": "integer", "description": "Maximum number of results (default: 100, max: 1000)", "default": 100, "maximum": 1000 } }, "required": ["query"]
- mcp_wikidata/tools.py:171-172 (handler)MCP tool dispatcher handler that invokes the WikidataClient's sparql_query method with tool arguments.elif name == "sparql_query": result = await self.client.sparql_query(**arguments)