export_collection
Export all documents from a specified collection in Typesense. Handles large datasets but may be memory-intensive for very large collections. Useful for data backup, migration, or analysis.
Instructions
Exports all documents from a specific collection.
Warning: This can be memory-intensive for very large collections.
Args:
ctx (Context): The MCP context.
collection_name (str): The name of the collection to export.
Returns:
list[dict] | str: A list of document dictionaries or an error message string.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| collection_name | Yes |
Input Schema (JSON Schema)
{
"properties": {
"collection_name": {
"title": "Collection Name",
"type": "string"
}
},
"required": [
"collection_name"
],
"title": "export_collectionArguments",
"type": "object"
}
Implementation Reference
- main.py:170-209 (handler)The handler function for the 'export_collection' MCP tool. It exports all documents from the specified Typesense collection using the export() method, parsing JSON lines into a list of dicts, with comprehensive error handling.@mcp.tool() async def export_collection(ctx: Context, collection_name: str) -> list[dict] | str: """ Exports all documents from a specific collection. Warning: This can be memory-intensive for very large collections. Args: ctx (Context): The MCP context. collection_name (str): The name of the collection to export. Returns: list[dict] | str: A list of document dictionaries or an error message string. """ if not collection_name: return "Error: collection_name parameter is required." documents = [] try: client: typesense.Client = ctx.request_context.lifespan_context.client # Check if collection exists first to give a clearer error _ = client.collections[collection_name].retrieve() # Check existence synchronously exported_lines = client.collections[collection_name].documents.export() for line in exported_lines: try: documents.append(json.loads(line)) except json.JSONDecodeError: print(f"Warning: Could not decode JSON line during export: {line}") # Decide whether to skip or raise an error continue return documents except typesense.exceptions.ObjectNotFound: return f"Error: Collection '{collection_name}' not found." except typesense.exceptions.TypesenseClientError as e: print(f"Error exporting collection '{collection_name}': {e}") return f"Error exporting collection '{collection_name}': {e}" except Exception as e: print(f"An unexpected error occurred while exporting collection '{collection_name}': {e}") return f"An unexpected error occurred: {e}"