ConceptNet MCP Server

Instructions

Implementation Reference

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

  Core handler function implementing the concept_query MCP tool. Handles parameter validation, ConceptNet API querying, response processing, error handling, and formats output in minimal or verbose modes.
  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/server.py:194-275 (registration)

  Registers the concept_query tool with the FastMCP server, providing detailed description and wrapper that delegates to the core handler.
  @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")
• cloudflare-workers/src/main.py:153-171 (schema)

  Explicit JSON schema definition for the concept_query tool input parameters in the Cloudflare Workers deployment.
  "concept_query": { "name": "concept_query", "description": "Advanced querying of ConceptNet with multi-parameter filtering. Returns minimal format (~96% smaller) by default or verbose format with full metadata when verbose=true.", "inputSchema": { "type": "object", "properties": { "start": {"type": "string", "description": "Start concept of relationships"}, "end": {"type": "string", "description": "End concept of relationships"}, "rel": {"type": "string", "description": "Relation type"}, "node": {"type": "string", "description": "Concept that must be start or end"}, "other": {"type": "string", "description": "Used with node parameter"}, "sources": {"type": "string", "description": "Filter by data source"}, "language": {"type": "string", "default": "en", "description": "Language filter"}, "limit_results": {"type": "boolean", "default": False, "description": "Limit to first page of results"}, "verbose": {"type": "boolean", "default": False, "description": "Return detailed format with full metadata (default: false for minimal format)"} }, "required": [] } },
• src/conceptnet_mcp/tools/concept_query.py:175-279 (helper)

  Helper function for validating input parameters and converting them to proper ConceptNet URI formats.
  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
• src/conceptnet_mcp/tools/concept_query.py:281-308 (helper)

  Helper function that constructs the QueryFilters object used for the ConceptNet API query.
  async def _build_query_filters( validated_params: Dict[str, Optional[str]], limit_results: bool, ctx: Context ) -> QueryFilters: """Build QueryFilters object from validated parameters.""" try: filters = QueryFilters( start=validated_params["start"], end=validated_params["end"], rel=validated_params["rel"], node=validated_params["node"], other=validated_params["other"], sources=validated_params["sources"], limit=20 if limit_results else 1000, offset=0 ) # Log the filters being applied specified_filters = filters.get_specified_filters() await ctx.info(f"Applied filters: {', '.join(specified_filters)}") return filters except Exception as e: raise MCPValidationError("query_filters", str(validated_params), f"Valid query combination: {e}")