Typesense MCP Server

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.

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}"