Guide · Vector Databases

MCP Tools for Milvus — schema-first collections, explicit load(), column-format insert, expr filtering, partition management

Milvus has three operational requirements that catch developers coming from Pinecone or Qdrant: collections must be loaded into memory before searches can run — collection.load() is a separate, explicit step after collection creation and after MCP server restart, and attempting to search an unloaded collection returns a "collection not loaded" error with no guidance in the error message about the load step; data is inserted in column format, not row format — you pass a list of field arrays where each array contains all values for one field ([[id1, id2], [vec1, vec2], [meta1, meta2]]), not a list of records where each record has all fields; and scalar filtering uses a string expression DSL, not a JSON filter object — expr="category == 'article' and score >= 0.5" uses Python-like syntax, not the MongoDB-style {"$eq": ...} operators used by Pinecone, Qdrant, and Chroma.

TL;DR

Connect: connections.connect(alias="default", host, port) or MilvusClient(uri, token) for Zilliz Cloud. Create collection: define CollectionSchema with FieldSchema list including INT64 primary key and FLOAT_VECTOR field. Build index: collection.create_index(field_name, index_params). Load: collection.load() — mandatory before search. Insert: collection.insert([[id_values], [vector_values], [scalar_values]]). Search: collection.search(data, anns_field, param, limit, expr="field > 0", output_fields). Health: GET /healthz → HTTP 200.

Connection and authentication

Milvus self-hosted accepts connections on port 19530 (gRPC, used by PyMilvus) and port 9091 (REST HTTP). The default credentials are root / Milvus. For Zilliz Cloud (managed Milvus), use the MilvusClient with the cluster endpoint URI and an API token. The PyMilvus connections.connect() API manages a global connection registry by alias — the alias "default" is used automatically by Collection objects that don't specify an alias.

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

# Self-hosted Milvus (gRPC port 19530)
connections.connect(
    alias="default",
    host=os.environ.get("MILVUS_HOST", "localhost"),
    port=os.environ.get("MILVUS_PORT", "19530"),
    user=os.environ.get("MILVUS_USER", "root"),
    password=os.environ.get("MILVUS_PASSWORD", "Milvus"),
)

# Zilliz Cloud — use MilvusClient with URI + token
from pymilvus import MilvusClient
zilliz_client = MilvusClient(
    uri=os.environ["ZILLIZ_URI"],    # e.g. https://in03-xyz.api.gcp-us-west1.zillizcloud.com
    token=os.environ["ZILLIZ_TOKEN"], # Zilliz API key (format: user:password or API key string)
)

# Verify connection
server_version = utility.get_server_version()
print(f"Milvus version: {server_version}")  # e.g. "v2.4.1"

# List collections to verify access
existing_collections = utility.list_collections()
print(f"Collections: {existing_collections}")

PyMilvus uses gRPC by default — the REST API at port 9091 is a separate interface primarily for monitoring and management. For MCP tools, always use PyMilvus (gRPC) or the official Go/Java SDKs rather than calling the REST API directly — the REST API has a different request format and not all operations are available via REST.

Schema-first collection design — FieldSchema and CollectionSchema

Milvus requires an explicit schema before creating a collection. Every schema must have exactly one primary key field (INT64 or VARCHAR) and exactly one vector field (FLOAT_VECTOR, BINARY_VECTOR, or FLOAT16_VECTOR). Additional scalar fields (INT64, FLOAT, VARCHAR, BOOL, JSON) are optional metadata. The schema cannot be changed after the collection is created — to add or remove fields you must drop and recreate the collection.

# Define collection schema
fields = [
    FieldSchema(
        name="id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=False,  # True = Milvus auto-generates IDs; False = you supply them
    ),
    FieldSchema(
        name="embedding",
        dtype=DataType.FLOAT_VECTOR,
        dim=1536,        # must match your embedding model's output dimension exactly
    ),
    FieldSchema(
        name="category",
        dtype=DataType.VARCHAR,
        max_length=256,  # required for VARCHAR fields — maximum byte length, not character count
    ),
    FieldSchema(
        name="score",
        dtype=DataType.FLOAT,
    ),
    FieldSchema(
        name="published_at",
        dtype=DataType.INT64,   # store timestamps as Unix epoch milliseconds
    ),
    FieldSchema(
        name="metadata",
        dtype=DataType.JSON,    # flexible key-value store for additional fields
    ),
]

schema = CollectionSchema(
    fields=fields,
    description="Document embeddings for semantic search",
    enable_dynamic_field=True,  # allow fields not in schema to be inserted (stored in _$meta)
)

collection = Collection(name="documents", schema=schema)

# Check if collection already exists before creating
if not utility.has_collection("documents"):
    collection = Collection(name="documents", schema=schema)
else:
    collection = Collection(name="documents")  # attach to existing collection

The enable_dynamic_field=True option allows inserting fields not declared in the schema — they are stored in an internal JSON column _$meta. Dynamic fields can be used in expr filters using the syntax json_contains(_$meta["key"], value). For frequently-filtered fields, declare them explicitly in the schema rather than relying on dynamic fields — explicitly declared fields are indexed and filter faster.

Index creation and mandatory load() — the two-step before search

After creating a collection and inserting data, two explicit steps are required before search: (1) create an index on the vector field, and (2) call collection.load(). Searching without an index uses brute-force scan (FLAT) which works but is slow for large collections. Searching without load() raises MilvusException: collection not loaded — this is the most common Milvus integration error and occurs because Milvus keeps collections unloaded (off-heap) by default to conserve memory.

# Step 1: Create vector index (choose based on dataset size and accuracy requirements)
index_params = {
    "metric_type": "COSINE",       # COSINE | IP (inner product) | L2
    "index_type": "HNSW",          # HNSW | IVF_FLAT | IVF_SQ8 | FLAT
    "params": {
        "M": 16,                   # HNSW: neighbors per node (8-64, default 16)
        "efConstruction": 200,     # HNSW: candidates during build (affects recall)
    },
}

# For IVF_FLAT (good for 1M–100M vectors):
# index_params = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 1024}}
# nlist = number of cluster centroids; √(num_vectors) is a common starting value

collection.create_index(
    field_name="embedding",
    index_params=index_params,
    index_name="embedding_hnsw",  # optional name for multi-index collections
)

# Step 2: Load collection into memory — REQUIRED before any search or query
# This is blocking by default — returns when loading is complete
collection.load()

# For large collections, loading is async; poll load state:
# from pymilvus import utility
# utility.wait_for_loading_complete("documents", timeout=120)
# or check: utility.load_state("documents")  # "Loaded" | "Loading" | "NotLoad"

# After MCP server restart, collection must be loaded again
# Add to your MCP server startup:
def ensure_collection_loaded(collection_name: str) -> Collection:
    col = Collection(collection_name)
    load_state = utility.load_state(collection_name)
    if load_state.name != "Loaded":
        col.load()
        utility.wait_for_loading_complete(collection_name, timeout=60)
    return col

The load() call loads all collection data into query node memory. For collections with billions of vectors, this can take minutes and consume significant RAM. Use collection.load(partition_names=["partition_a"]) to load only specific partitions if memory is constrained. The collection remains loaded until collection.release() is called or the Milvus server restarts — at which point all collections return to unloaded state.

Column-format insert — not row format

Milvus inserts data in column-oriented format: each element of the insert array corresponds to all values for one field. This is the opposite of row-oriented databases where each element is one record. Passing row-format data (list of dicts) to the PyMilvus insert() method raises a type error. The column format matches how Milvus stores data internally and avoids row-to-column transposition overhead.

# WRONG — row format raises an error
row_data = [
    {"id": 1, "embedding": [0.1, ...], "category": "article"},
    {"id": 2, "embedding": [0.2, ...], "category": "blog"},
]
# collection.insert(row_data)  # TypeError: wrong format

# CORRECT — column format: one list per field, all lists same length
ids =        [1, 2, 3]
embeddings = [[0.1, 0.2, ...], [0.3, 0.4, ...], [0.5, 0.6, ...]]
categories = ["article", "blog", "article"]
scores =     [0.9, 0.7, 0.8]

# Field order in the insert list must match schema field order (excluding auto_id fields)
insert_result = collection.insert([ids, embeddings, categories, scores])
print(f"Inserted {insert_result.insert_count} records")
# insert_result.primary_keys contains the IDs of inserted records

# After inserting, call flush() to persist to disk (optional — auto-flushed on collection close)
# collection.flush()  # blocks until all inserted data is sealed and queryable

# More readable: use dict format (supported in newer PyMilvus versions)
# collection.insert({"id": ids, "embedding": embeddings, "category": categories})

Newly inserted data is available for search immediately (via a growing segment that uses FLAT scan), but is not indexed by the HNSW or IVF index until the segment is sealed and the background indexing completes. For predictable search performance in high-insert workloads, call collection.flush() after large batches to force segment sealing and trigger indexing.

Expr filtering — string DSL, not JSON operators

Milvus uses a string expression language for scalar filtering, similar to Python conditionals. Supported operators: ==, !=, >, >=, <, <=, in, not in, like (% wildcard), and, or, not. JSON field access: metadata["key"] == "value". Array operations: array_contains(tags, "python").

# Scalar filtering with expr string (passed to search or query)
search_params = {
    "metric_type": "COSINE",
    "params": {
        "ef": 64,      # HNSW search-time candidates (higher = more accurate, slower)
                       # must be >= limit; recommended: 2-4x limit
    },
}

results = collection.search(
    data=[query_vector],            # list of query vectors (one per search)
    anns_field="embedding",         # name of the vector field to search
    param=search_params,
    limit=10,                       # top-K results
    expr="category == 'article' and score >= 0.5",  # string DSL filter
    output_fields=["category", "score", "published_at"],  # scalar fields to return
)

# More complex expr examples:
# "category in ['article', 'blog'] and score between 0.3 and 0.9"
# "metadata[\"lang\"] == 'en' or metadata[\"lang\"] == 'de'"
# "not (category == 'spam')"
# "array_contains(tags, 'machine-learning')"
# "published_at >= 1700000000000"  # Unix milliseconds

# Query (no vector — scalar-only filter, returns all matching records)
query_results = collection.query(
    expr="category == 'article' and score >= 0.8",
    output_fields=["id", "category", "score"],
    limit=1000,
)

Milvus's expr DSL does not support IS NULL or IS NOT NULL directly — use the JSON field workaround: json_contains_all(metadata, ["optional_key"]) to check presence. For fields declared in the schema (not dynamic fields), null values are not permitted — all scalar fields are required on insert unless you set a default_value in the FieldSchema. Set default_value=0 or default_value="" for optional scalar fields to avoid insert errors.

Partitions — logical sub-division within a collection

Milvus partitions are named sub-divisions within a collection that allow targeted loading and searching. The default partition "_default" exists in every collection. Partitions are useful for multi-tenant scenarios (one partition per tenant) or time-based data (one partition per month). Loading and searching can be scoped to specific partitions to reduce memory footprint and improve search speed.

# Create partitions
collection.create_partition("2024_q1")
collection.create_partition("2024_q2")
collection.create_partition("tenant_abc")

# Insert into a specific partition
collection.insert(
    data=[ids, embeddings, categories, scores],
    partition_name="2024_q1",
)

# Load only specific partitions (reduces memory usage)
collection.load(partition_names=["2024_q1", "2024_q2"])

# Search within specific partitions
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=10,
    partition_names=["2024_q1"],  # search only Q1 data
)

# List partitions and their record counts
partitions = collection.partitions
for p in partitions:
    print(f"Partition: {p.name}, rows: {p.num_entities}")

Partition key support (Milvus 2.2.9+) allows automatic routing of records to partitions based on a scalar field value — declare is_partition_key=True on a VARCHAR or INT64 field in the schema, and Milvus will hash values to partitions automatically. This replaces manual partition management for tenant isolation use cases and removes the need to specify partition_name on every insert.

Health probe — /healthz and collection load state

Milvus exposes GET /healthz on port 9091 (REST) for liveness — it returns HTTP 200 when the server is healthy. For collection-level readiness, check the load state via PyMilvus or the REST API. A collection in NotLoad state will fail all search operations even if /healthz returns 200.

import fetch from 'node-fetch';

// HTTP REST health check (Milvus port 9091)
async function checkMilvusHealth(
  milvusHost: string,
  collectionName: string
): Promise<{ alive: boolean; collectionLoaded: boolean }> {
  // Liveness check via REST
  const healthRes = await fetch(`http://${milvusHost}:9091/healthz`);
  if (!healthRes.ok) return { alive: false, collectionLoaded: false };

  // Collection load state via REST (Milvus 2.x REST API)
  const stateRes = await fetch(
    `http://${milvusHost}:9091/v1/vector/collections/get_load_state`,
    {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ collection_name: collectionName }),
    }
  );
  if (!stateRes.ok) return { alive: true, collectionLoaded: false };
  const stateData = await stateRes.json();
  // stateData.data.loadState: "LoadStateLoaded" | "LoadStateLoading" | "LoadStateNotExist" | "LoadStateNotLoad"

  return {
    alive: true,
    collectionLoaded: stateData.data?.loadState === 'LoadStateLoaded',
  };
}

// Using PyMilvus for the same check:
# from pymilvus import utility
# load_state = utility.load_state("documents")  # returns LoadState enum
# is_loaded = (load_state == LoadState.Loaded)

Build a startup probe for your MCP server that checks both /healthz liveness and collection load state. If the collection is not loaded (LoadStateNotLoad), call collection.load() automatically as part of startup. This handles Milvus server restarts transparently without requiring manual operator intervention to reload collections before your MCP server can serve search requests.