MCP Wikidata Server

Instructions

Implementation Reference

• mcp_wikidata/wikidata_client.py:193-244 (handler)

  The main handler function that executes the get_relations tool logic. It constructs a SPARQL query based on relation_type (incoming or outgoing), executes it via sparql_query, processes the results into a structured list of relations, and returns them.

  async def get_relations(
      self,
      entity_id: str,
      relation_type: str = "outgoing",
      property_filter: Optional[List[str]] = None,
      limit: int = 20
  ) -> Dict[str, Any]:
      if relation_type == "outgoing":
          query = f"""
          SELECT ?property ?propertyLabel ?target ?targetLabel WHERE {{
            wd:{entity_id} ?property ?target .
            ?prop wikibase:directClaim ?property .
            SERVICE wikibase:label {{ bd:serviceParam wikibase:language "en" . }}
          }}
          LIMIT {limit}
          """
      elif relation_type == "incoming":
          query = f"""
          SELECT ?property ?propertyLabel ?source ?sourceLabel WHERE {{
            ?source ?property wd:{entity_id} .
            ?prop wikibase:directClaim ?property .
            SERVICE wikibase:label {{ bd:serviceParam wikibase:language "en" . }}
          }}
          LIMIT {limit}
          """
      else:
          raise ValueError(f"Invalid relation_type: {relation_type}")
  
      result = await self.sparql_query(query)
      
      relations = []
      for binding in result.get("results", {}).get("bindings", []):
          relation = {
              "property": binding.get("property", {}).get("value", "").split("/")[-1],
              "property_label": binding.get("propertyLabel", {}).get("value", ""),
              "direction": relation_type
          }
          
          if relation_type == "outgoing":
              relation["target"] = {
                  "id": binding.get("target", {}).get("value", "").split("/")[-1],
                  "label": binding.get("targetLabel", {}).get("value", "")
              }
          else:
              relation["source"] = {
                  "id": binding.get("source", {}).get("value", "").split("/")[-1],
                  "label": binding.get("sourceLabel", {}).get("value", "")
              }
          
          relations.append(relation)
  
      return {"relations": relations}

• mcp_wikidata/tools.py:103-132 (schema)

  The tool schema definition including input validation (inputSchema) for get_relations, specifying parameters like entity_id, relation_type, property_filter, and limit.

  Tool(
      name="get_relations",
      description="Get relations of a Wikidata entity",
      inputSchema={
          "type": "object",
          "properties": {
              "entity_id": {
                  "type": "string",
                  "description": "Wikidata entity ID"
              },
              "relation_type": {
                  "type": "string",
                  "description": "Type of relations to retrieve",
                  "enum": ["incoming", "outgoing", "all"],
                  "default": "outgoing"
              },
              "property_filter": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Filter by specific properties"
              },
              "limit": {
                  "type": "integer",
                  "description": "Maximum number of relations",
                  "default": 20,
                  "maximum": 100
              }
          },
          "required": ["entity_id"]
      }

• mcp_wikidata/tools.py:173-174 (registration)

  Registration in the call_tool dispatcher method, which routes the tool call to the WikidataClient's get_relations method.

  elif name == "get_relations":
      result = await self.client.get_relations(**arguments)

• mcp_wikidata/tools.py:163-165 (registration)

  The get_tool_definitions method registers the get_relations tool by including it in the returned list of Tool objects, making it available to the MCP server.

      ]
  
  async def call_tool(self, name: str, arguments: Dict[str, Any]) -> List[TextContent]: