R2R FastMCP Server

Instructions

Implementation Reference

• server.py:457-637 (handler)

  The core handler function that executes the 'search' tool logic, performing R2R retrieval search with configurable parameters and formatting results for LLM.
  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)

  FastMCP decorator registering the search function as an MCP tool with annotations.
  @mcp.tool( annotations={ "title": "R2R Search", "readOnlyHint": True, "idempotentHint": True, "openWorldHint": True, } )
• server.py:457-536 (schema)

  Input/output schema defined by function parameters, types, defaults, and comprehensive docstring documentation.
  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") """
• server.py:29-98 (helper)

  Helper function to format various types of R2R search results into a readable string for LLMs.
  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 to load preset configurations for different search scenarios.
  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()