R2R FastMCP Server

Instructions

Implementation Reference

• server.py:457-637 (handler)

  The main handler function implementing the 'search' MCP tool. It validates parameters, configures search settings based on presets and explicit params, executes the R2R search, formats results, and returns them as a string for LLM consumption.
  async def search( query: str, ctx: Context, preset: str = "default", use_semantic_search: bool = True, use_hybrid_search: bool = False, use_graph_search: bool = True, limit: int = 10, kg_search_type: Literal["local", "global"] = "local", semantic_weight: float = 5.0, full_text_weight: float = 1.0, full_text_limit: int = 200, rrf_k: int = 50, search_strategy: str | None = "rag_fusion", include_web_search: bool = False, ) -> str: """ Perform comprehensive search on R2R knowledge base with full parameter control. This tool supports semantic search, hybrid search (semantic + full-text), knowledge graph search, and web search. Use presets for common scenarios or customize all parameters manually. Args: query: The search query to find relevant documents. Required. preset: Preset configuration for common use cases. Options: - "default": Basic semantic search, 10 results - "development": Hybrid search optimized for code development, 15 results - "refactoring": Hybrid + graph search for code refactoring, 20 results - "debug": Minimal graph search for debugging, 5 results - "research": Comprehensive search with global graph, 30 results - "production": Balanced hybrid search for production, 10 results use_semantic_search: Enable semantic/vector search (default: True) use_hybrid_search: Enable hybrid search combining semantic and full-text search (default: False) use_graph_search: Enable knowledge graph search for entity/relationship discovery (default: False) limit: Maximum number of results to return. Must be between 1 and 100 (default: 10) kg_search_type: Knowledge graph search type. "local" for local context, "global" for broader connections (default: "local") semantic_weight: Weight for semantic search in hybrid mode. Must be between 0.0 and 10.0 (default: 5.0) full_text_weight: Weight for full-text search in hybrid mode. Must be between 0.0 and 10.0 (default: 1.0) full_text_limit: Maximum full-text results to consider in hybrid search. Must be between 1 and 1000 (default: 200) rrf_k: Reciprocal Rank Fusion parameter for hybrid search. Must be between 1 and 100 (default: 50) search_strategy: Advanced search strategy (e.g., "hyde", "rag_fusion"). Optional. include_web_search: Include web search results from the internet (default: False) Returns: Formatted search results including: - Vector search results (chunks) - Graph search results (entities, relationships, communities) - Web search results (if enabled) - Document search results (local documents with chunks) Examples: # Simple search with default settings search("What is machine learning?") # Development preset for code search search("async function implementation", preset="development") # Custom hybrid search search( "API documentation", use_hybrid_search=True, semantic_weight=7.0, limit=20 ) # Research with knowledge graph search("neural network architectures", preset="research") """ await ctx.info(f"Starting search query: {query}, preset: {preset}") try: # Validate parameters validate_limit(limit) validate_semantic_weight(semantic_weight) validate_full_text_weight(full_text_weight) validate_full_text_limit(full_text_limit) validate_rrf_k(rrf_k) if use_graph_search: validate_kg_search_type(kg_search_type) await ctx.report_progress(progress=10, total=100, message="Initializing client") client = R2RClient(base_url=R2R_BASE_URL) if API_KEY: client.set_api_key(API_KEY) # Get preset configuration and merge with explicit parameters preset_config = get_search_preset_config(preset) # Apply preset values, but allow explicit parameters to override # For boolean flags: if preset enables it, use it unless explicitly disabled # For numeric: use preset if value is default, otherwise use explicit value final_use_hybrid = ( use_hybrid_search if preset == "default" else (use_hybrid_search or preset_config.get("use_hybrid_search", False)) ) final_use_graph = ( use_graph_search if preset == "default" else (use_graph_search or preset_config.get("use_graph_search", False)) ) search_settings: dict[str, Any] = { "use_semantic_search": use_semantic_search, "limit": limit, } # Apply hybrid search settings if final_use_hybrid: search_settings["use_hybrid_search"] = True hybrid_config = preset_config.get("hybrid_settings", {}) search_settings["hybrid_settings"] = { "semantic_weight": semantic_weight if semantic_weight != 5.0 or preset == "default" else hybrid_config.get("semantic_weight", 5.0), "full_text_weight": full_text_weight if full_text_weight != 1.0 or preset == "default" else hybrid_config.get("full_text_weight", 1.0), "full_text_limit": full_text_limit if full_text_limit != 200 or preset == "default" else hybrid_config.get("full_text_limit", 200), "rrf_k": rrf_k if rrf_k != 50 or preset == "default" else hybrid_config.get("rrf_k", 50), } await ctx.info("Hybrid search enabled") # Apply graph search settings if final_use_graph: kg_type = ( kg_search_type if kg_search_type != "local" or preset == "default" else preset_config.get("kg_search_type", "local") ) search_settings["graph_search_settings"] = { "use_graph_search": True, "kg_search_type": kg_type, } await ctx.info(f"Knowledge graph search enabled (type: {kg_type})") # Apply search strategy if provided if search_strategy: search_settings["search_strategy"] = search_strategy await ctx.info(f"Search strategy: {search_strategy}") await ctx.report_progress(progress=30, total=100, message="Executing search") search_response = client.retrieval.search( query=query, search_settings=search_settings ) await ctx.report_progress(progress=80, total=100, message="Formatting results") formatted = format_search_results_for_llm(search_response.results) await ctx.report_progress(progress=100, total=100, message="Complete") await ctx.info( f"Search completed successfully, returned {len(formatted)} chars" ) return formatted except ValueError as e: await ctx.error(f"Validation error: {e!s}") raise except Exception as e: await ctx.error(f"Search failed: {e!s}") raise
• server.py:449-456 (registration)

  The FastMCP decorator that registers the 'search' function as an MCP tool with metadata annotations indicating it's read-only, idempotent, and suitable for open-world use.
  @mcp.tool( annotations={ "title": "R2R Search", "readOnlyHint": True, "idempotentHint": True, "openWorldHint": True, } )
• server.py:29-98 (helper)

  Helper function that formats the raw R2R search results (chunk, graph, web, document) into a readable string with sections for LLM input.
  def format_search_results_for_llm(results: Any) -> str: """ Format R2R search results for LLM consumption. Aggregates 4 types of results: - Chunk search (vector search) - Graph search (knowledge graph entities/relationships) - Web search (internet results) - Document search (local documents with chunks) """ lines = [] # 1) Chunk search if results.chunk_search_results: lines.append("Vector Search Results:") for c in results.chunk_search_results: lines.append(f"Source ID [{id_to_shorthand(c.id)}]:") lines.append(c.text or "") # 2) Graph search if results.graph_search_results: lines.append("Graph Search Results:") for g in results.graph_search_results: lines.append(f"Source ID [{id_to_shorthand(g.id)}]:") if hasattr(g.content, "summary"): lines.append(f"Community Name: {g.content.name}") lines.append(f"ID: {g.content.id}") lines.append(f"Summary: {g.content.summary}") elif hasattr(g.content, "name") and hasattr(g.content, "description"): lines.append(f"Entity Name: {g.content.name}") lines.append(f"Description: {g.content.description}") elif ( hasattr(g.content, "subject") and hasattr(g.content, "predicate") and hasattr(g.content, "object") ): rel = f"{g.content.subject}-{g.content.predicate}-{g.content.object}" lines.append(f"Relationship: {rel}") # 3) Web search if results.web_search_results: lines.append("Web Search Results:") for w in results.web_search_results: lines.append(f"Source ID [{id_to_shorthand(w.id)}]:") lines.append(f"Title: {w.title}") lines.append(f"Link: {w.link}") lines.append(f"Snippet: {w.snippet}") # 4) Local context docs if results.document_search_results: lines.append("Local Context Documents:") for doc_result in results.document_search_results: doc_title = doc_result.title or "Untitled Document" doc_id = doc_result.id summary = doc_result.summary lines.append(f"Full Document ID: {doc_id}") lines.append(f"Shortened Document ID: {id_to_shorthand(doc_id)}") lines.append(f"Document Title: {doc_title}") if summary: lines.append(f"Summary: {summary}") if doc_result.chunks: for chunk in doc_result.chunks: lines.append( f"\nChunk ID {id_to_shorthand(chunk['id'])}:\n{chunk['text']}" ) result = "\n".join(lines) return result
• server.py:124-200 (helper)

  Helper function providing preset configurations for different search scenarios (default, development, refactoring, etc.), used to set default search parameters.
  def get_search_preset_config(preset: str) -> dict[str, Any]: """ Get search configuration for a preset. Args: preset: Preset name (default, development, refactoring, debug, research, production) Returns: Dictionary with search settings """ presets = { "default": { "use_semantic_search": True, "use_hybrid_search": False, "use_graph_search": False, "limit": 10, }, "development": { "use_semantic_search": True, "use_hybrid_search": True, "use_graph_search": False, "limit": 15, "hybrid_settings": { "semantic_weight": 5.0, "full_text_weight": 1.0, "full_text_limit": 200, "rrf_k": 50, }, }, "refactoring": { "use_semantic_search": True, "use_hybrid_search": True, "use_graph_search": True, "limit": 20, "kg_search_type": "local", "hybrid_settings": { "semantic_weight": 7.0, "full_text_weight": 3.0, "full_text_limit": 300, "rrf_k": 50, }, }, "debug": { "use_semantic_search": True, "use_hybrid_search": False, "use_graph_search": True, "limit": 5, "kg_search_type": "local", }, "research": { "use_semantic_search": True, "use_hybrid_search": True, "use_graph_search": True, "limit": 30, "kg_search_type": "global", "hybrid_settings": { "semantic_weight": 6.0, "full_text_weight": 2.0, "full_text_limit": 400, "rrf_k": 60, }, }, "production": { "use_semantic_search": True, "use_hybrid_search": True, "use_graph_search": False, "limit": 10, "hybrid_settings": { "semantic_weight": 5.0, "full_text_weight": 1.0, "full_text_limit": 200, "rrf_k": 50, }, }, } return presets.get(preset.lower(), presets["default"]).copy()
• server.py:320-361 (helper)

  Group of validation helper functions for search parameters: validate_limit, validate_temperature, validate_semantic_weight, etc.
  def validate_limit(limit: int) -> None: """Validate limit parameter (1-100).""" if not 1 <= limit <= 100: raise ValueError("limit must be between 1 and 100") def validate_temperature(temperature: float) -> None: """Validate temperature parameter (0.0-1.0).""" if not 0.0 <= temperature <= 1.0: raise ValueError("temperature must be between 0.0 and 1.0") def validate_semantic_weight(weight: float) -> None: """Validate semantic weight parameter (0.0-10.0).""" if not 0.0 <= weight <= 10.0: raise ValueError("semantic_weight must be between 0.0 and 10.0") def validate_full_text_weight(weight: float) -> None: """Validate full text weight parameter (0.0-10.0).""" if not 0.0 <= weight <= 10.0: raise ValueError("full_text_weight must be between 0.0 and 10.0") def validate_full_text_limit(limit: int) -> None: """Validate full text limit parameter (1-1000).""" if not 1 <= limit <= 1000: raise ValueError("full_text_limit must be between 1 and 1000") def validate_rrf_k(k: int) -> None: """Validate RRF k parameter (1-100).""" if not 1 <= k <= 100: raise ValueError("rrf_k must be between 1 and 100") def validate_kg_search_type(kg_search_type: str) -> None: """Validate knowledge graph search type.""" if kg_search_type not in ["local", "global"]: raise ValueError("kg_search_type must be 'local' or 'global'")