localvectordb_server.mcp.server module

LocalVectorDB MCP Server (stdio-based)

Provides Model Context Protocol server for LocalVectorDB, enabling LLMs to interact with vector databases through a unified tool interface.

class localvectordb_server.mcp.server.MCPManager(config: MCPConfig)

Bases: object

Simple manager for MCP database operations using VectorDB factory

__init__(config: MCPConfig)

async get_database(name: str): Get database instance using factory pattern - auto-detects local/remote

async list_databases() → List[str]: List available databases

async create_database(name: str, metadata_schema: Dict[str, Any] | None = None, **kwargs): Create a new database (write mode only)

async delete_database(name: str): Delete a database (write mode only)

async cleanup(): Cleanup resources on shutdown

localvectordb_server.mcp.server.lifespan(mcp): Lifespan context manager for MCP server initialization and cleanup

localvectordb_server.mcp.server.register_tool(name: str, read_only: bool = True): Decorator to register tools in the registry

localvectordb_server.mcp.server.register_mcp_tool(func): Helper to register a function as an MCP tool with proper metadata

async localvectordb_server.mcp.server.list_databases() → Dict[str, Any]

List all available vector databases

Returns:: Dictionary with database names and count

async localvectordb_server.mcp.server.get_database_info(database_name: str) → Dict[str, Any]

Get detailed information about a specific database

Parameters:: database_name – Name of the database
Returns:: Database statistics and configuration

async localvectordb_server.mcp.server.query_database(database_name: str, query: str, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'context', 'enriched', 'sections'] | None = None, search_level: Literal['chunks', 'sections', 'documents'] = 'chunks', k: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1)])] = 10, score_threshold: float = 0.0, filters: Dict | None = None, vector_weight: float = 0.5, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: str = 'auto') → Dict[str, Any]

Search a database using vector, keyword, or hybrid search

If the database was created with a persisted default reranker, results are automatically reranked by it (cross-encoder re-scoring of the candidate pool).

Parameters:

database_name – Name of the database to search
query – Search query text
search_type – Type of search (vector, keyword, hybrid)
return_type – Return documents, chunks, context, enriched, or sections. Defaults to the unit search_level searched: documents for ‘chunks’, sections for ‘sections’. Set return_type=’documents’ with search_level=’sections’ to rank documents by their best section. (‘sections’ requires a database created with hierarchical_embeddings=True)
search_level – Which index to query — ‘chunks’ (default), ‘sections’, or ‘documents’ (‘sections’/’documents’ require a database created with hierarchical_embeddings=True)
k – Number of results to return
score_threshold – Minimum score threshold
filters – Metadata filters (MongoDB-style). Filter fields must exist in the database’s metadata schema; unknown fields or operators are rejected with an error
vector_weight – Weight for vector search in hybrid mode (0-1)
context_window – Size of the assembled context for return_type=’context’/’enriched’, measured in context_unit (chunk count when ‘chunks’, otherwise a token/word/character budget)
context_unit – Unit for context_window (‘chunks’, ‘tokens’, ‘words’, ‘characters’)
context_truncate – Hard-truncate the assembled context to exactly the budget (only applies with a non-chunk context_unit)
semantic_dedup_threshold – Threshold for semantic deduplication
document_scoring_method – Method for scoring documents

Returns:

Search results with scores and metadata

async localvectordb_server.mcp.server.find_related_documents(database_name: str, document_id: str, k: int = 5, score_threshold: float = 0.0, filters: Dict | None = None) → Dict[str, Any]

Find documents related to a given document (nearest neighbours by embedding)

Returns the documents most similar to document_id using document-level embeddings, sorted by descending similarity. The reference document itself is excluded.

Parameters:

database_name – Name of the database
document_id – Reference document to find neighbours for
k – Maximum number of related documents to return
score_threshold – Minimum similarity score to include
filters – Metadata filters (MongoDB-style) applied to candidate documents. Filter fields must exist in the database’s metadata schema; unknown fields or operators are rejected with an error

Returns:

Related documents with similarity scores and metadata

async localvectordb_server.mcp.server.filter_documents(database_name: str, filters: Dict[str, Any], limit: int = 100, offset: int = 0) → Dict[str, Any]

Filter documents by metadata

Parameters:

database_name – Name of the database
filters – Metadata filters (MongoDB-style)
limit – Maximum number of results
offset – Number of results to skip

Returns:

Filtered documents

async localvectordb_server.mcp.server.get_document(database_name: str, document_id: str, chunk: str | None = None, char_range: str | None = None, line_range: str | None = None, section: str | None = None, outline: bool = False) → Dict[str, Any]

Retrieve a document, or a selected portion of it, by ID

By default the whole document is returned. The selection arguments below return a part of it instead and are mutually exclusive (pass at most one).

Parameters:

database_name – Name of the database
document_id – ID of the document
chunk – Return stored chunk(s) by 0-based index or inclusive range ‘M:N’ (e.g. ‘3’ or ‘2:5’). Adds a ‘chunks’ list (index/content/position).
char_range – Return a character slice ‘M:N’ (0-based, end-exclusive)
line_range – Return a line range ‘M:N’ (1-based, inclusive)
section – Return the section whose Markdown heading matches this name (case-insensitive)
outline – Return the document’s section outline (headings, levels, lines) as an ‘outline’ list instead of content

Returns:

Document metadata plus the requested content/chunks/outline. The ‘mode’ field reports which selection produced the result.

async localvectordb_server.mcp.server.check_documents_exist(database_name: str, document_ids: List[str]) → Dict[str, Any]

Check if documents exist in the database

Parameters:

database_name – Name of the database
document_ids – List of document IDs to check

Returns:

Dictionary mapping document IDs to existence status

async localvectordb_server.mcp.server.get_metadata_schema(database_name: str) → Dict[str, Any]

Get the metadata schema for a database

Parameters:: database_name – Name of the database
Returns:: Metadata schema definition

async localvectordb_server.mcp.server.grep_documents(database_name: str, pattern: str, regex: bool = False, ignore_case: bool = False, whole_word: bool = False, context: int = 0, before_context: int | None = None, after_context: int | None = None, prefix: str | None = None, filters: Dict[str, Any] | None = None, max_count: int | None = None, limit: int = 50) → Dict[str, Any]

Lexical, line-oriented search over document content – like grep.

This is exact/regex substring matching, deliberately separate from query_database: query_database does ranked semantic/keyword retrieval, while grep_documents finds literal or regex matches and reports where they are (document id, 1-based line number, column span, and optional surrounding lines). Use it alongside vector and keyword search when you know a precise string or pattern to look for. Matches are returned in document-id then line order, not by relevance.

Parameters:

database_name – Name of the database to search
pattern – Text to search for – a literal substring by default, or a regular expression when regex=True
regex – Treat pattern as a regular expression
ignore_case – Case-insensitive matching
whole_word – Match only whole words (word boundaries around the pattern)
context – Number of context lines to include both before and after each match
before_context – Context lines before each match (overrides context)
after_context – Context lines after each match (overrides context)
prefix – Restrict the search to documents whose id starts with this prefix (e.g. ‘docs/’ when using path-like ids)
filters – Metadata filters (MongoDB-style) restricting which documents are searched. Filter fields must exist in the database’s metadata schema.
max_count – Stop after this many matches per document
limit – Maximum total matches to return across all documents (default 50). Keep this modest – a broad pattern can otherwise return a large, token-heavy result. ‘truncated’ is True when the cap was hit.

Returns:

Matches with line numbers, column spans, and optional context lines

async localvectordb_server.mcp.server.list_prefixes(database_name: str, prefix: str = '', delimiter: str = '/') → Dict[str, Any]

List the immediate children of a document-id prefix, S3-style.

Treats path-like document ids (e.g. ‘docs/reports/q1’) as a virtual folder hierarchy and rolls documents up to their first segment beneath prefix. There are no real directories – only ids that share a prefix. ‘prefixes’ holds the virtual sub-folders and ‘documents’ holds the leaf documents that live directly at this level; pass a returned prefix’s ‘path’ back as prefix to descend one level.

Parameters:

database_name – Name of the database
prefix – Id prefix to list beneath (empty lists the top level)
delimiter – Virtual path separator used to split ids into segments (default ‘/’)

Returns:

The immediate sub-prefixes and leaf documents at this level, with counts

async localvectordb_server.mcp.server.get_system_info() → Dict[str, Any]

Get system information and configuration

Returns:: System version, configuration, and status

async localvectordb_server.mcp.server.create_database(name: str, metadata_schema: Dict[str, Any] | None = None, embedding_provider: str | None = None, embedding_model: str | None = None, chunking_method: str | None = None, chunk_size: int | None = None, chunk_overlap: int | None = None) → Dict[str, Any]

Create a new vector database

Parameters:

name – Database name
metadata_schema – Schema for document metadata (field_name -> type or config dict). Valid field types: text, integer, real, boolean, date, json. Only declared fields can be written or filtered on
embedding_provider – Provider for embeddings (e.g., “ollama”, “openai”)
embedding_model – Model name for embeddings
chunking_method – Method for chunking documents
chunk_size – Maximum chunk size
chunk_overlap – Overlap between chunks

Returns:

Database configuration and status

async localvectordb_server.mcp.server.delete_database(name: str) → Dict[str, Any]

Delete a vector database

Parameters:: name – Database name to delete
Returns:: Deletion status

Insert or update documents in the database

Upserting over an EXISTING id replaces the whole document: content and metadata alike, so metadata not re-sent is cleared. To change content while preserving existing metadata, use update_document (which merges) or patch_document (which does not touch metadata).

Parameters:

database_name – Name of the database
documents – Document(s) to upsert
metadata – Metadata for documents. Every field must be declared in the database’s metadata schema (metadata_schema at create_database time, or update_metadata_schema later); undeclared fields are rejected
ids – Optional document IDs. Each id may appear at most once per call; reusing an id in a later call updates that document.
batch_size – Batch size for processing
similarity_threshold – Threshold for similarity detection

Returns:

Document IDs and operation status

async localvectordb_server.mcp.server.update_document(database_name: str, document_id: str, content: str | None = None, metadata: Dict[str, Any] | None = None) → Dict[str, Any]

Update a document’s content and/or metadata

Metadata is MERGED: fields you send are updated, fields you omit keep their stored values, and a content-only update preserves all metadata. This is the opposite of upsert_documents, which replaces the whole document.

Parameters:

database_name – Name of the database
document_id – ID of the document to update
content – New content (optional)
metadata – New or updated metadata (optional). Every field must be declared in the database’s metadata schema; undeclared fields are rejected

Returns:

Update status

async localvectordb_server.mcp.server.patch_document(database_name: str, document_id: str, old_string: str, new_string: str, count: int = 1, expect_hash: str | None = None) → Dict[str, Any]

Edit a document in place by replacing exact text, without re-sending the whole document. This mirrors the find/replace contract of a code-editing tool: the old_string must occur exactly count times or the edit fails, so it is unambiguous and never silently corrupts the untouched remainder.

Parameters:

database_name – Name of the database
document_id – ID of the document to patch
old_string – Exact text to find (must match count times)
new_string – Text to replace it with
count – Number of expected occurrences of old_string (default 1)
expect_hash – Optional content_hash precondition; if it does not match the stored document the patch fails with a conflict instead of clobbering a concurrent edit.

Returns:

Patch status including new_hash and ops_applied.

async localvectordb_server.mcp.server.delete_document(database_name: str, document_id: str) → Dict[str, Any]

Delete a document from the database

Parameters:

database_name – Name of the database
document_id – ID of the document to delete

Returns:

Deletion status

async localvectordb_server.mcp.server.update_metadata_schema(database_name: str, metadata_schema: Dict[str, Any]) → Dict[str, Any]

Update the metadata schema for a database

Parameters:

database_name – Name of the database
metadata_schema – New metadata schema definition (field_name -> type or config dict). Valid field types: text, integer, real, boolean, date, json

Returns:

Update status

async localvectordb_server.mcp.server.run_mcp_server(mode: str = 'read-only'): Run the stdio MCP server