index_multiple_documents
Index multiple documents in a Typesense collection by specifying an action (create, upsert, or update). Processes documents in a batch and returns results for each, including success or error details.
Instructions
Indexes (creates, upserts, or updates) multiple documents in a batch.
Args:
ctx (Context): The MCP context.
collection_name (str): The name of the collection.
documents (list[dict]): A list of document dictionaries to index.
action (str): The import action ('create', 'upsert', 'update'). Defaults to 'upsert'.
Returns:
list[dict] | str: A list of result dictionaries (one per document) or an error message string.
Each result dict typically looks like {'success': true/false, 'error': '...', 'document': {...}}.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| action | No | upsert | |
| collection_name | Yes | ||
| documents | Yes |
Input Schema (JSON Schema)
{
"properties": {
"action": {
"default": "upsert",
"title": "Action",
"type": "string"
},
"collection_name": {
"title": "Collection Name",
"type": "string"
},
"documents": {
"items": {
"additionalProperties": true,
"type": "object"
},
"title": "Documents",
"type": "array"
}
},
"required": [
"collection_name",
"documents"
],
"title": "index_multiple_documentsArguments",
"type": "object"
}
Implementation Reference
- main.py:649-701 (handler)The handler function for the 'index_multiple_documents' tool, decorated with @mcp.tool() for registration. It performs batch import of documents into a Typesense collection using the specified action (create, upsert, update), parsing JSONL responses and handling various errors.@mcp.tool() async def index_multiple_documents( ctx: Context, collection_name: str, documents: list[dict], action: str = 'upsert' ) -> list[dict] | str: """ Indexes (creates, upserts, or updates) multiple documents in a batch. Args: ctx (Context): The MCP context. collection_name (str): The name of the collection. documents (list[dict]): A list of document dictionaries to index. action (str): The import action ('create', 'upsert', 'update'). Defaults to 'upsert'. Returns: list[dict] | str: A list of result dictionaries (one per document) or an error message string. Each result dict typically looks like {'success': true/false, 'error': '...', 'document': {...}}. """ if not collection_name: return "Error: collection_name parameter is required." if not isinstance(documents, list) or not documents: return "Error: documents parameter must be a non-empty list of dictionaries." if action not in ['create', 'upsert', 'update']: return "Error: action parameter must be one of 'create', 'upsert', 'update'." results = [] try: client: typesense.Client = ctx.request_context.lifespan_context.client # NOTE: Assuming import_ is *sync* based on observed pattern # Also, the response is JSONL strings, not directly awaitable objects import_response_lines = client.collections[collection_name].documents.import_(documents, {'action': action}) # Response is JSON Lines (string per result), parse each line # NOTE: Changed to normal 'for' loop for line in import_response_lines: try: results.append(json.loads(line)) except json.JSONDecodeError: print(f"Warning: Could not decode JSON line from import response: {line}") results.append({'success': False, 'error': 'Failed to decode response line', 'raw_line': line}) return results except typesense.exceptions.ObjectNotFound: return f"Error: Collection '{collection_name}' not found." except typesense.exceptions.RequestMalformed as e: return f"Error: Malformed bulk import request for collection '{collection_name}'. Check document structures. Details: {e}" except typesense.exceptions.TypesenseClientError as e: print(f"Error during bulk import to '{collection_name}': {e}") return f"Error during bulk import to '{collection_name}': {e}" except Exception as e: print(f"An unexpected error occurred during bulk import to '{collection_name}': {e}") return f"An unexpected error occurred during bulk import: {e}"
- main.py:656-668 (schema)The docstring provides the input schema (arguments with types and descriptions) and output format for the tool.""" Indexes (creates, upserts, or updates) multiple documents in a batch. Args: ctx (Context): The MCP context. collection_name (str): The name of the collection. documents (list[dict]): A list of document dictionaries to index. action (str): The import action ('create', 'upsert', 'update'). Defaults to 'upsert'. Returns: list[dict] | str: A list of result dictionaries (one per document) or an error message string. Each result dict typically looks like {'success': true/false, 'error': '...', 'document': {...}}. """
- main.py:649-649 (registration)The @mcp.tool() decorator registers the function as an MCP tool.@mcp.tool()