server.py

Overview

server.py implements a microservice control plane (MCP) server for RAGFlow, a Retrieval-Augmented Generation (RAG) backend system that facilitates querying large document datasets. The server exposes an API interface that allows clients to query datasets and documents through retrieval requests. It supports multiple transport protocols including Server-Sent Events (SSE) and Streamable HTTP, enabling real-time and asynchronous data streaming.

This file handles:

Classes and Functions

LaunchMode (StrEnum)

Enum defining the launch mode of the server.

Transport (StrEnum)

Enum representing supported transport protocols.

RAGFlowConnector

Connector class that communicates with the RAGFlow backend API. It manages caching of datasets and document metadata and facilitates retrieval queries.

Properties

Initialization

def __init__(self, base_url: str, version="v1")

connector = RAGFlowConnector(base_url="http://127.0.0.1:9380")

Methods

RAGFlowCtx

Simple context wrapper holding a RAGFlowConnector instance.

sse_lifespan(server: Server) -> AsyncIterator[dict]

Async context manager for the lifecycle of the legacy SSE application. Initializes the RAGFlowConnector context and logs startup/shutdown events.

with_api_key(required=True)

Decorator factory for API key injection and validation on endpoint handlers.

@with_api_key(required=True)
async def list_tools(*, connector):
    ...

MCP Server Endpoints

list_tools(*, connector) -> list[types.Tool]

call_tool(name: str, arguments: dict, *, connector) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]

create_starlette_app() -> Starlette

Creates a Starlette ASGI application configured with:

Returns the configured Starlette app instance.

main(...)

CLI entrypoint using click to parse options and environment variables. Sets global configuration variables, validates mode constraints, prints banner and status, and runs the server via uvicorn.

Implementation Details and Algorithms

Interaction with Other Components

Usage Examples

Starting the Server in Self-Host Mode

uv run mcp/server/server.py --host=127.0.0.1 --port=9382 --base-url=http://127.0.0.1:9380 --mode=self-host --api-key=ragflow-xxxxx

Calling the Retrieval Tool Programmatically

from mcp.server.server import RAGFlowConnector

connector = RAGFlowConnector(base_url="http://127.0.0.1:9380")
connector.bind_api_key("ragflow-xxxxx")

results = connector.retrieval(
    dataset_ids=["dataset1"],
    question="What is the capital of France?",
    page=1,
    page_size=10,
)

for content in results:
    print(content.text)

Mermaid Class Diagram

classDiagram
    class RAGFlowConnector {
        -_MAX_DATASET_CACHE: int
        -_MAX_DOCUMENT_CACHE: int
        -_CACHE_TTL: int
        -_dataset_metadata_cache: OrderedDict
        -_document_metadata_cache: OrderedDict
        +__init__(base_url: str, version="v1")
        +bind_api_key(api_key: str)
        +list_datasets(page: int, page_size: int, orderby: str, desc: bool, id: str, name: str) str
        +retrieval(dataset_ids, document_ids, question, page, page_size, similarity_threshold, vector_similarity_weight, top_k, rerank_id, keyword, force_refresh)
        -_post(path, json, stream, files)
        -_get(path, params, json)
        -_is_cache_valid(ts)
        -_get_expiry_timestamp()
        -_get_cached_dataset_metadata(dataset_id)
        -_set_cached_dataset_metadata(dataset_id, metadata)
        -_get_cached_document_metadata_by_dataset(dataset_id)
        -_set_cached_document_metadata_by_dataset(dataset_id, doc_id_meta_list)
        -_get_document_metadata_cache(dataset_ids, force_refresh)
        -_map_chunk_fields(chunk_data, dataset_cache, document_cache)
    }

    class RAGFlowCtx {
        +__init__(connector: RAGFlowConnector)
        -conn: RAGFlowConnector
    }

    RAGFlowCtx --> RAGFlowConnector

Summary

server.py provides a robust MCP server interface to the RAGFlow retrieval backend, supporting multi-tenant and self-host modes with flexible transport protocols. It implements efficient caching strategies, rich metadata enrichment, and secure API key management to enable scalable and performant retrieval operations. The integration with Starlette and uvicorn allows asynchronous and extensible server deployment.

This file is a core part of the RAGFlow system enabling external clients to query, retrieve, and interact with large document datasets in a real-time, scalable manner.