sparql_query
Execute SPARQL queries to retrieve structured data from Wikidata, supporting JSON, CSV, and XML formats with configurable result limits.
Instructions
Execute a SPARQL query against Wikidata
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SPARQL query | |
| format | No | Response format | json |
| limit | No | Maximum number of results (default: 100, max: 1000) |
Implementation Reference
- mcp_wikidata/wikidata_client.py:145-192 (handler)The main handler function that executes the SPARQL query logic: modifies query if no LIMIT, sends HTTP GET to Wikidata SPARQL endpoint, parses JSON or returns text, handles timeouts, rate limits, errors.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)Registers the sparql_query tool in MCP by defining its Tool object with name, description, and inputSchema in the WikidataTools.get_tool_definitions() method.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:171-172 (handler)Dispatches execution of sparql_query tool calls to the underlying WikidataClient instance.elif name == "sparql_query": result = await self.client.sparql_query(**arguments)
- mcp_wikidata/tools.py:80-101 (schema)Defines the JSON schema for input parameters of the sparql_query tool: query (required string), optional format (json/csv/xml), optional limit (int).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"] }