Production Vector Search MCP Integration: Pinecone and Qdrant

graph TD
    Agent[AI Agent: tools/call semantic_code_search] --> MCP[Vector Search MCP Server]
    
    subgraph Query Processing Pipeline
        MCP --> Embed[Dense Embedding: text-embedding-3-small]
        MCP --> Tokenize[Sparse Lexical: BM25 / SPLADE]
    end
    
    Embed --> DenseQuery[Dense Vector Search]
    Tokenize --> SparseQuery[Sparse Keyword Search]
    
    DenseQuery --> VectorDB[(Pinecone / Qdrant Cluster)]
    SparseQuery --> VectorDB
    
    VectorDB --> RRF[Reciprocal Rank Fusion RRF Re-ranker]
    RRF --> TokenBudget[Token Budget Truncator: Max 4,000 Tokens]
    TokenBudget --> Markdown[Token-Efficient Markdown Serializer]
    Markdown --> Agent

{
  "name": "enterprise_vector_search",
  "description": "Hybrid semantic and keyword search across internal engineering documentation, API references, and codebase indexing.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "Natural language query or exact keyword identifier."
      },
      "namespace": {
        "type": "string",
        "enum": ["engineering_docs", "codebase_index", "incident_postmortems"],
        "description": "Target document corpus."
      },
      "max_chunks": {
        "type": "integer",
        "default": 5,
        "maximum": 10,
        "description": "Number of top-ranked chunks to return."
      },
      "filter_tags": {
        "type": "array",
        "items": { "type": "string" },
        "description": "Optional metadata filters (e.g. ['typescript', 'v2'])."
      }
    },
    "required": ["query", "namespace"]
  }
}

import os
import tiktoken
from typing import List, Optional
from mcp.server.fastmcp import FastMCP
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Prefetch,
    FusionQuery,
    Fusion,
    Filter,
    FieldCondition,
    MatchAny
)
from openai import OpenAI

# 1. Initialize Clients & Encoders
mcp = FastMCP("enterprise-vector-search")
qdrant = QdrantClient(url=os.getenv("QDRANT_URL"), api_key=os.getenv("QDRANT_API_KEY"))
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
tokenizer = tiktoken.get_encoding("cl100k_base")

TOKEN_BUDGET_LIMIT = 3500

# 2. Hybrid Retrieval with Reciprocal Rank Fusion
@mcp.tool()
async def enterprise_vector_search(
    query: str,
    namespace: str,
    max_chunks: int = 5,
    filter_tags: Optional[List[str]] = None
) -> str:
    """
    Search indexed knowledge base using dense vector embeddings and BM25 sparse matching.
    """
    # Generate Dense Embedding
    dense_resp = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    dense_vector = dense_resp.data[0].embedding

    # Build Qdrant Metadata Filters
    query_filter = None
    if filter_tags:
        query_filter = Filter(
            must=[FieldCondition(key="tags", match=MatchAny(any=filter_tags))]
        )

    # Execute Hybrid RRF Query in Qdrant
    results = qdrant.query_points(
        collection_name=namespace,
        prefetch=[
            Prefetch(
                query=dense_vector,
                using="dense",
                limit=max_chunks * 2,
                filter=query_filter
            ),
            # Sparse prefetch assumes SPLADE or BM25 vector indexed in 'sparse'
            Prefetch(
                query={"indices": [10, 45, 99], "values": [0.8, 0.4, 0.9]}, 
                using="sparse",
                limit=max_chunks * 2,
                filter=query_filter
            )
        ],
        query=FusionQuery(fusion=Fusion.RRF),
        limit=max_chunks
    )

    # 3. Format and Truncate Response for LLM Context Budget
    formatted_output = []
    total_tokens = 0

    for idx, point in enumerate(results.points, start=1):
        doc_id = point.id
        score = round(point.score, 4)
        text = point.payload.get("text", "").strip()
        source_url = point.payload.get("source_url", "internal://unknown")

        chunk_header = f"### [Source {idx}]: {source_url} (RRF Score: {score})\n"
        chunk_body = f"{text}\n\n"
        chunk_combined = chunk_header + chunk_body

        chunk_tokens = len(tokenizer.encode(chunk_combined))

        # Enforce hard token ceiling
        if total_tokens + chunk_tokens > TOKEN_BUDGET_LIMIT:
            formatted_output.append("> *[Notice: Additional lower-ranked search chunks truncated to protect context budget]*\n")
            break

        formatted_output.append(chunk_combined)
        total_tokens += chunk_tokens

    if not formatted_output:
        return "No relevant documentation found matching the search criteria."

    return "".join(formatted_output)

if __name__ == "__main__":
    mcp.run(transport="stdio")

{
  "jsonrpc": "2.0",
  "id": "vec-search-401",
  "method": "tools/call",
  "params": {
    "name": "enterprise_vector_search",
    "arguments": {
      "query": "How to configure mTLS for gRPC internal services",
      "namespace": "engineering_docs",
      "max_chunks": 2,
      "filter_tags": ["security", "networking"]
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": "vec-search-401",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "### [Source 1]: https://docs.enterprise.internal/security/mtls-grpc (RRF Score: 0.9234)\nTo enable mTLS on internal gRPC services, provision certificates via Vault PKI and set ServerCredentials.createSslContext().\n\n### [Source 2]: https://docs.enterprise.internal/networking/envoy-sidecar (RRF Score: 0.8841)\nEnvoy sidecars automatically terminate internal mTLS when configured with strict SPIFFE identities."
      }
    ]
  }
}