rag-mcp-server-builder

IBM RAG MCP Server Builder

Purpose

Expert guidance for building Model Context Protocol (MCP) servers that expose RAG capabilities (ingestion, retrieval, hybrid search) as tools consumable by IBM Bob, Claude, and other MCP-compatible AI assistants.

IBM Cloud Product Coverage

Rules

• Use fastmcp or the mcp Python SDK for server implementation
• SSE transport: serve at GET /sse and POST /messages/
• stdio transport: pipe JSON-RPC messages through stdin/stdout
• Tool names: use snake_case (e.g. ingest_documents, search_documents, hybrid_search)
• Always validate inputs with Pydantic v2 before calling IBM services
• IBM Code Engine: expose port 8080; set SSE_TRANSPORT=true env var

Scope

• RAG ingestion MCP tools (from IBM COS to Milvus/OpenSearch)
• RAG retrieval MCP tools (vector search, keyword search, hybrid search)
• Embedding generation MCP tools (IBM watsonx.ai)
• SSE transport server for remote deployment on IBM Code Engine
• stdio transport server for local IBM Bob integration

Procedure

Phase 1: MCP Server Structure (SSE)

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from fastapi import FastAPI

server = Server("rag-retrieval-server")
app = FastAPI()
sse = SseServerTransport("/messages/")

@app.get("/sse")
async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await server.run(streams[0], streams[1], server.create_initialization_options())

Phase 2: Register Retrieval Tool

from mcp.types import Tool, TextContent

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="search_documents",
            description="Semantic search over IBM watsonx.data vector store using IBM watsonx.ai embeddings",
            inputSchema={
                "type": "object",
                "properties": {
                    "query":      {"type": "string", "description": "Natural language search query"},
                    "index_name": {"type": "string", "description": "Vector index / collection name"},
                    "top_k":      {"type": "integer", "default": 5},
                    "search_type":{"type": "string", "enum": ["vector", "keyword", "hybrid"], "default": "hybrid"},
                },
                "required": ["query", "index_name"],
            },
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "search_documents":
        results = await do_search(
            query=arguments["query"],
            index_name=arguments["index_name"],
            top_k=arguments.get("top_k", 5),
            search_type=arguments.get("search_type", "hybrid"),
        )
        return [TextContent(type="text", text=json.dumps(results, indent=2))]