ConceptNet MCP Server

Instructions

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