upsert_document
Insert or update a document in a Typesense collection using its unique ID. Ensures data accuracy and management in MCP-driven search operations.
Instructions
Upserts (creates or updates) a single document in a specific collection.
Args:
ctx (Context): The MCP context.
collection_name (str): The name of the collection.
document (dict): The document data to upsert (must include an 'id' field).
Returns:
dict | str: The upserted document dictionary or an error message string.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| collection_name | Yes | ||
| document | Yes |
Implementation Reference
- main.py:614-647 (handler)The main handler function for the 'upsert_document' tool. It is decorated with @mcp.tool(), which registers it as an MCP tool. The function upserts a document into a Typesense collection using the client from the application context, with input validation and comprehensive error handling.
@mcp.tool() async def upsert_document(ctx: Context, collection_name: str, document: dict) -> dict | str: """ Upserts (creates or updates) a single document in a specific collection. Args: ctx (Context): The MCP context. collection_name (str): The name of the collection. document (dict): The document data to upsert (must include an 'id' field). Returns: dict | str: The upserted document dictionary or an error message string. """ if not collection_name: return "Error: collection_name parameter is required." if not isinstance(document, dict) or 'id' not in document: return "Error: document parameter must be a dictionary and include an 'id' field." try: client: typesense.Client = ctx.request_context.lifespan_context.client # NOTE: Assuming upsert is *sync* based on observed pattern upserted_doc = client.collections[collection_name].documents.upsert(document) return upserted_doc except typesense.exceptions.ObjectNotFound: return f"Error: Collection '{collection_name}' not found." except typesense.exceptions.RequestMalformed as e: return f"Error: Malformed upsert document request for collection '{collection_name}'. Check document structure against schema. Details: {e}" except typesense.exceptions.TypesenseClientError as e: print(f"Error upserting document in '{collection_name}': {e}") return f"Error upserting document in '{collection_name}': {e}" except Exception as e: print(f"An unexpected error occurred while upserting document in '{collection_name}': {e}") return f"An unexpected error occurred: {e}"