Typesense MCP Server

Instructions

Creates a new collection with the provided schema.

Args:
    ctx (Context): The MCP context.
    schema (dict): The collection schema dictionary (must include 'name' and 'fields').

Returns:
    dict | str: The created collection schema dictionary or an error message string.

Implementation Reference

• main.py:455-484 (handler)

  The handler function for the 'create_collection' tool. It is decorated with @mcp.tool(), which registers the tool in the FastMCP server. The function validates the schema input, accesses the Typesense client from the context, and creates the collection asynchronously, handling various exceptions.

  @mcp.tool()
  async def create_collection(ctx: Context, schema: dict) -> dict | str:
      """
      Creates a new collection with the provided schema.
  
      Args:
          ctx (Context): The MCP context.
          schema (dict): The collection schema dictionary (must include 'name' and 'fields').
  
      Returns:
          dict | str: The created collection schema dictionary or an error message string.
      """
      if not isinstance(schema, dict) or 'name' not in schema or 'fields' not in schema:
          return "Error: Invalid schema provided. Must be a dictionary with 'name' and 'fields' keys."
  
      try:
          client: typesense.Client = ctx.request_context.lifespan_context.client
          # Assuming create is async based on library structure
          created_collection = await client.collections.create(schema)
          return created_collection
      except typesense.exceptions.ObjectAlreadyExists:
          return f"Error: Collection '{schema.get('name')}' already exists."
      except typesense.exceptions.RequestMalformed as e:
           return f"Error: Malformed create collection request. Check schema format. Details: {e}"
      except typesense.exceptions.TypesenseClientError as e:
          print(f"Error creating collection '{schema.get('name')}': {e}")
          return f"Error creating collection '{schema.get('name')}': {e}"
      except Exception as e:
          print(f"An unexpected error occurred while creating collection '{schema.get('name')}': {e}")
          return f"An unexpected error occurred: {e}"