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

TableJSON Schema

Name	Required	Default
`action`	No	upsert
`collection_name`	Yes
`documents`	Yes

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()
```

Typesense MCP Server