ConceptNet MCP Server

Instructions

Advanced querying of ConceptNet with sophisticated multi-parameter filtering.

This tool provides powerful filtering capabilities for exploring ConceptNet's
knowledge graph. You can combine multiple filters to find specific types of
relationships and concepts with precision.

Features:
- Multi-parameter filtering (start, end, relation, node, sources)
- Complex relationship discovery and analysis
- Comprehensive result processing and enhancement
- Query optimization and performance metrics
- Format control: minimal (~96% smaller) vs verbose (full metadata)

Format Options:
- verbose=false (default): Returns minimal format optimized for LLM consumption
- verbose=true: Returns comprehensive format with full ConceptNet metadata
- Backward compatibility maintained with existing tools

Filter Parameters:
- start: Start concept of relationships (e.g., "dog", "/c/en/dog")
- end: End concept of relationships (e.g., "animal", "/c/en/animal")
- rel: Relation type (e.g., "IsA", "/r/IsA")
- node: Concept that must be either start or end of edges
- other: Used with 'node' to find relationships between two specific concepts
- sources: Filter by data source (e.g., "wordnet", "/s/activity/omcs")

Use this when you need:
- Precise relationship filtering and discovery
- Complex queries with multiple constraints
- Analysis of specific relationship types
- Targeted exploration of concept connections

Implementation Reference

• src/conceptnet_mcp/tools/concept_query.py:43-173 (handler)

  The primary handler function implementing the core logic of the 'concept_query' tool. Handles parameter validation, ConceptNet API querying, response processing, error handling, and returns either minimal or verbose formatted results.

  async def concept_query(
      ctx: Context,
      start: Optional[str] = None,
      end: Optional[str] = None,
      rel: Optional[str] = None,
      node: Optional[str] = None,
      other: Optional[str] = None,
      sources: Optional[str] = None,
      language: str = "en",
      limit_results: bool = False,
      verbose: bool = False
  ) -> Dict[str, Any]:
      """
      Advanced querying of ConceptNet with multiple filters.
      
      This tool provides sophisticated filtering capabilities for exploring
      ConceptNet's knowledge graph. You can combine multiple filters to find
      specific types of relationships and concepts. By default, returns a
      minimal format optimized for LLM consumption.
      
      Args:
          start: Start concept URI or term (e.g., "dog", "/c/en/dog")
          end: End concept URI or term (e.g., "animal", "/c/en/animal")
          rel: Relation type (e.g., "IsA", "/r/IsA")
          node: Concept that must be either start or end of edges
          other: Used with 'node' to find relationships between two specific concepts
          sources: Filter by data source (e.g., "wordnet", "/s/activity/omcs")
          language: Language filter for concepts (default: "en")
          limit_results: If True, limits to 20 results for quick queries (default: False)
          verbose: If True, returns detailed format with full metadata (default: False)
          
      Returns:
          Query results with relationships grouped by type (minimal format) or
          comprehensive data with full metadata (verbose format).
          
      Examples:
          - concept_query(start="dog", rel="IsA") -> Minimal format with grouped relationships
          - concept_query(start="dog", rel="IsA", verbose=True) -> Full detailed format
          - concept_query(node="car", other="transportation") -> Car-transportation relationships
      """
      start_time = datetime.now(timezone.utc)
      
      try:
          # Log the incoming request
          await ctx.info(f"Starting advanced ConceptNet query with multiple filters")
          
          # 1. Parameter validation and conversion
          await ctx.info("Validating and processing query parameters...")
          validated_params = await _validate_and_convert_parameters(
              start, end, rel, node, other, sources, language, ctx
          )
          
          # 2. Build QueryFilters object
          filters = await _build_query_filters(validated_params, limit_results, ctx)
          
          # 3. Query ConceptNet API
          await ctx.info(f"Executing ConceptNet query: {filters}")
          
          async with ConceptNetClient() as client:
              try:
                  response = await client.query_concepts(
                      filters=filters,
                      get_all_pages=not limit_results,
                      target_language=language if language != "en" else None
                  )
              except ConceptNetAPIError as e:
                  return _create_api_error_response(validated_params, str(e), start_time)
          
          # 4. Process and enhance response
          await ctx.info("Processing and enhancing query results...")
          processor = ResponseProcessor(default_language=language)
          
          # Process edges with same-language filtering by default
          edges = response.get("edges", [])
          # Apply same-language filtering by default
          filtered_edges = processor.filter_by_language(edges, language, require_both=True)
          processed_edges = processor.process_edge_list(filtered_edges, target_language=language)
          
          if len(filtered_edges) != len(edges):
              await ctx.info(f"Applied same-language filtering ({language}): {len(edges)} → {len(filtered_edges)} edges")
          
          # 5. Return appropriate format based on verbose parameter
          if verbose:
              # Return detailed format with full metadata (existing behavior)
              enhanced_response = await _create_enhanced_query_response(
                  processed_edges, response, validated_params, filters,
                  language, limit_results, start_time, ctx
              )
              
              total_edges = len(processed_edges)
              await ctx.info(f"Successfully completed query with {total_edges} results (verbose format)")
              
              return enhanced_response
          else:
              # Return minimal format optimized for LLMs
              # Create a mock processed response for the minimal formatter
              mock_response = {"edges": processed_edges}
              
              # Determine concept term for minimal response (use first non-None parameter)
              concept_term = (
                  validated_params.get("start") or
                  validated_params.get("end") or
                  validated_params.get("node") or
                  "query_results"
              )
              
              # Extract just the term part if it's a URI
              if concept_term and concept_term.startswith('/c/'):
                  parts = concept_term.split('/')
                  if len(parts) >= 4:
                      concept_term = parts[3]  # Extract term from /c/lang/term
              
              minimal_response = processor.create_minimal_concept_response(
                  mock_response, concept_term
              )
              
              total_relationships = minimal_response.get("summary", {}).get("total_relationships", 0)
              await ctx.info(f"Successfully completed query with {total_relationships} relationships (minimal format)")
              
              return minimal_response
          
      except MCPValidationError as e:
          return _create_validation_error_response(e, start_time)
          
      except ConceptNetAPIError as e:
          return _create_api_error_response({}, str(e), start_time)
          
      except Exception as e:
          logger.error(f"Unexpected error in concept_query: {e}")
          return _create_unexpected_error_response(str(e), start_time)

• src/conceptnet_mcp/models/query.py:13-150 (schema)

  Pydantic BaseModel defining the input schema and validation for query filters used by the concept_query tool handler.

  class QueryFilters(BaseModel):
      """
      Query parameters for ConceptNet API advanced search.
      
      This model encapsulates all the filtering options available for the ConceptNet
      query endpoint, providing validation and sensible defaults for pagination.
      """
      
      start: Optional[str] = Field(
          default=None,
          description="Filter by start concept URI (e.g., '/c/en/dog')"
      )
      end: Optional[str] = Field(
          default=None,
          description="Filter by end concept URI (e.g., '/c/en/animal')"
      )
      rel: Optional[str] = Field(
          default=None,
          description="Filter by relation type URI (e.g., '/r/IsA')"
      )
      node: Optional[str] = Field(
          default=None,
          description="Filter by concept that appears as either start or end"
      )
      other: Optional[str] = Field(
          default=None,
          description="Filter by concept different from the 'node' parameter"
      )
      sources: Optional[str] = Field(
          default=None,
          description="Filter by source dataset or contributor"
      )
      limit: int = Field(
          default=20,
          description="Maximum number of results to return",
          ge=1,
          le=1000
      )
      offset: int = Field(
          default=0,
          description="Number of results to skip for pagination",
          ge=0
      )
      
      @field_validator('start', 'end', 'node', 'other')
      @classmethod
      def validate_concept_uris(cls, v: Optional[str]) -> Optional[str]:
          """Validate concept URIs start with '/c/'."""
          if v is not None and not v.startswith('/c/'):
              raise ValueError("Concept URIs must start with '/c/'")
          return v
      
      @field_validator('rel')
      @classmethod
      def validate_relation_uri(cls, v: Optional[str]) -> Optional[str]:
          """Validate relation URIs start with '/r/'."""
          if v is not None and not v.startswith('/r/'):
              raise ValueError("Relation URIs must start with '/r/'")
          return v
      
      @model_validator(mode='after')
      def validate_query_logic(self) -> Self:
          """Validate that the query parameters make logical sense together."""
          # If 'other' is specified, 'node' must also be specified
          if self.other is not None and self.node is None:
              raise ValueError("'other' parameter requires 'node' parameter to be specified")
          
          # 'node' and 'other' cannot be the same
          if self.node is not None and self.other is not None and self.node == self.other:
              raise ValueError("'node' and 'other' parameters cannot be the same")
          
          # At least one filter parameter should be specified for meaningful queries
          filter_params = [self.start, self.end, self.rel, self.node, self.sources]
          if all(param is None for param in filter_params):
              raise ValueError("At least one filter parameter must be specified")
          
          return self
      
      def to_query_params(self) -> dict[str, str]:
          """
          Convert this model to a dictionary suitable for HTTP query parameters.
          
          Returns:
              Dictionary of non-None parameters as strings
          """
          params = {}
          
          if self.start is not None:
              params['start'] = self.start
          if self.end is not None:
              params['end'] = self.end
          if self.rel is not None:
              params['rel'] = self.rel
          if self.node is not None:
              params['node'] = self.node
          if self.other is not None:
              params['other'] = self.other
          if self.sources is not None:
              params['sources'] = self.sources
          
          params['limit'] = str(self.limit)
          params['offset'] = str(self.offset)
          
          return params
      
      def get_specified_filters(self) -> Set[str]:
          """
          Get the names of all filter parameters that have been specified.
          
          Returns:
              Set of parameter names that are not None
          """
          specified = set()
          if self.start is not None:
              specified.add('start')
          if self.end is not None:
              specified.add('end')
          if self.rel is not None:
              specified.add('rel')
          if self.node is not None:
              specified.add('node')
          if self.other is not None:
              specified.add('other')
          if self.sources is not None:
              specified.add('sources')
          return specified
      
      def __str__(self) -> str:
          """Return a human-readable string representation."""
          filters = self.get_specified_filters()
          return f"QueryFilters({', '.join(filters)}, limit={self.limit}, offset={self.offset})"
      
      def __repr__(self) -> str:
          """Return a detailed string representation for debugging."""
          return (f"QueryFilters(start='{self.start}', end='{self.end}', rel='{self.rel}', "
                  f"node='{self.node}', other='{self.other}', sources='{self.sources}', "
                  f"limit={self.limit}, offset={self.offset})")

• src/conceptnet_mcp/server.py:194-275 (registration)

  FastMCP tool registration decorator and wrapper function that registers 'concept_query' and delegates to the core handler implementation.

  @mcp.tool(
      name="concept_query",
      description="""
      Advanced querying of ConceptNet with sophisticated multi-parameter filtering.
      
      This tool provides powerful filtering capabilities for exploring ConceptNet's
      knowledge graph. You can combine multiple filters to find specific types of
      relationships and concepts with precision.
      
      Features:
      - Multi-parameter filtering (start, end, relation, node, sources)
      - Complex relationship discovery and analysis
      - Comprehensive result processing and enhancement
      - Query optimization and performance metrics
      - Format control: minimal (~96% smaller) vs verbose (full metadata)
      
      Format Options:
      - verbose=false (default): Returns minimal format optimized for LLM consumption
      - verbose=true: Returns comprehensive format with full ConceptNet metadata
      - Backward compatibility maintained with existing tools
      
      Filter Parameters:
      - start: Start concept of relationships (e.g., "dog", "/c/en/dog")
      - end: End concept of relationships (e.g., "animal", "/c/en/animal")
      - rel: Relation type (e.g., "IsA", "/r/IsA")
      - node: Concept that must be either start or end of edges
      - other: Used with 'node' to find relationships between two specific concepts
      - sources: Filter by data source (e.g., "wordnet", "/s/activity/omcs")
      
      Use this when you need:
      - Precise relationship filtering and discovery
      - Complex queries with multiple constraints
      - Analysis of specific relationship types
      - Targeted exploration of concept connections
      """,
      tags={"conceptnet", "query", "filtering", "advanced", "relationships"}
  )
  async def concept_query_tool(
      ctx: Context,
      start: Optional[str] = None,
      end: Optional[str] = None,
      rel: Optional[str] = None,
      node: Optional[str] = None,
      other: Optional[str] = None,
      sources: Optional[str] = None,
      language: str = "en",
      limit_results: bool = False,
      verbose: bool = False
  ) -> Dict[str, Any]:
      """
      MCP tool wrapper for advanced concept querying functionality.
      
      Args:
          start: Start concept URI or term (e.g., "dog", "/c/en/dog")
          end: End concept URI or term (e.g., "animal", "/c/en/animal")
          rel: Relation type (e.g., "IsA", "/r/IsA")
          node: Concept that must be either start or end of edges
          other: Used with 'node' to find relationships between two specific concepts
          sources: Filter by data source (e.g., "wordnet", "/s/activity/omcs")
          language: Language filter for concepts (default: "en")
          limit_results: If True, limits to 20 results for quick queries (default: False)
          verbose: If True, returns detailed format with full metadata (default: False)
          
      Returns:
          Query results with relationships grouped by type (minimal format) or comprehensive data with full metadata (verbose format)
      """
      try:
          return await concept_query(
              ctx=ctx,
              start=start,
              end=end,
              rel=rel,
              node=node,
              other=other,
              sources=sources,
              language=language,
              limit_results=limit_results,
              verbose=verbose
          )
      except Exception as e:
          return await handle_server_error(e, "concept_query")

• src/conceptnet_mcp/tools/concept_query.py:175-279 (helper)

  Helper function for validating input parameters and converting text terms to proper ConceptNet URIs.

  async def _validate_and_convert_parameters(
      start: Optional[str],
      end: Optional[str], 
      rel: Optional[str],
      node: Optional[str],
      other: Optional[str],
      sources: Optional[str],
      language: str,
      ctx: Context
  ) -> Dict[str, Optional[str]]:
      """Validate and convert all query parameters to proper ConceptNet URIs."""
      
      # Validate at least one filter is provided (not None and not empty string)
      filter_params = [start, end, rel, node, sources]
      if all(param is None or param == '' for param in filter_params):
          raise MCPValidationError(
              "query_parameters",
              "all_none_or_empty",
              "At least one filter parameter (start, end, rel, node, sources) must be provided"
          )
      
      # Validate 'other' requires 'node'
      if other and not node:
          raise MCPValidationError(
              "other", 
              other, 
              "'other' parameter requires 'node' parameter to be set"
          )
      
      # Validate language
      if not validate_language_code(language):
          await ctx.warning(f"Language code '{language}' may not be supported by ConceptNet")
      
      validated = {}
      
      # Convert concept parameters to URIs if needed
      for param_name, param_value in [("start", start), ("end", end), ("node", node), ("other", other)]:
          if param_value is not None:
              # Check for empty strings - these should be validation errors
              if param_value == '':
                  raise MCPValidationError(param_name, param_value, "Non-empty string")
              
              if param_value.startswith('/c/'):
                  # Already a URI, validate format
                  parts = param_value.split('/')
                  if len(parts) < 4:
                      raise InvalidConceptURIError(param_value, "/c/language/term")
                  validated[param_name] = param_value
              else:
                  # Convert text to URI
                  try:
                      validated[param_name] = create_concept_uri(param_value, language)
                      await ctx.debug(f"Converted {param_name}: '{param_value}' -> '{validated[param_name]}'")
                  except Exception as e:
                      raise MCPValidationError(param_name, param_value, f"Valid concept term or URI: {e}")
          else:
              validated[param_name] = None
      
      # Convert relation parameter if needed
      if rel is not None:
          # Check for empty strings
          if rel == '':
              raise MCPValidationError("rel", rel, "Non-empty string")
              
          if rel.startswith('/r/'):
              # Already a relation URI
              validated["rel"] = rel
          else:
              # Convert text to relation URI
              # Handle common relation names and convert to proper format
              normalized_rel = rel.title().replace(' ', '').replace('_', '')
              validated["rel"] = f"/r/{normalized_rel}"
              await ctx.debug(f"Converted relation: '{rel}' -> '{validated['rel']}'")
      else:
          validated["rel"] = None
      
      # Handle sources parameter
      if sources is not None:
          # Check for empty strings
          if sources == '':
              raise MCPValidationError("sources", sources, "Non-empty string")
              
          if sources.startswith('/s/'):
              # Already a source URI
              validated["sources"] = sources
          else:
              # Convert text to source URI format (common sources)
              source_mappings = {
                  "wordnet": "/s/resource/wordnet/rdf/3.1",
                  "dbpedia": "/s/resource/dbpedia/2015/en",
                  "omcs": "/s/activity/omcs",
                  "conceptnet": "/s/resource/conceptnet/5.7"
              }
              
              if sources.lower() in source_mappings:
                  validated["sources"] = source_mappings[sources.lower()]
                  await ctx.debug(f"Converted source: '{sources}' -> '{validated['sources']}'")
              else:
                  # Assume it's a custom source pattern
                  validated["sources"] = sources if sources.startswith('/s/') else f"/s/{sources}"
      else:
          validated["sources"] = None
      
      return validated