MCP Wikidata Server

Instructions

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"]
  }