copy_data
Insert data into Vertica tables using the COPY command to transfer rows from external sources into database storage.
Instructions
Copy data into a Vertica table using COPY command.
Args:
ctx: FastMCP context for progress reporting and logging
schema: vertica schema to execute the copy against
table: Target table name
data: List of rows to insert
Returns:
Status message indicating success or failureInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| schema | Yes | ||
| table | Yes | ||
| data | Yes |
Implementation Reference
- src/mcp_vertica/mcp.py:308-365 (handler)The 'copy_data' tool handler function. It is registered via the @mcp.tool() decorator. Takes schema, table, and data (list of lists), converts data to CSV, and uses Vertica's COPY command to insert the data efficiently. Includes permission checks and proper resource management.
@mcp.tool() async def copy_data( ctx: Context, schema: str, table: str, data: List[List[Any]], ) -> str: """Copy data into a Vertica table using COPY command. Args: ctx: FastMCP context for progress reporting and logging schema: vertica schema to execute the copy against table: Target table name data: List of rows to insert Returns: Status message indicating success or failure """ await ctx.info(f"Copying {len(data)} rows to table: {table}") # Get or create connection manager manager = await get_or_create_manager(ctx) if not manager: return "Error: Failed to initialize database connection. Check configuration." # Check operation permissions if not manager.is_operation_allowed(schema, OperationType.INSERT): error_msg = f"INSERT operation not allowed for database {schema}" await ctx.error(error_msg) return error_msg conn = None cursor = None try: conn = manager.get_connection() cursor = conn.cursor() # Convert data to CSV string output = io.StringIO() writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL) writer.writerows(data) output.seek(0) # Create COPY command copy_query = f"""COPY {table} FROM STDIN DELIMITER ',' ENCLOSED BY '\"'""" cursor.copy(copy_query, output.getvalue()) conn.commit() success_msg = f"Successfully copied {len(data)} rows to {table}" await ctx.info(success_msg) return success_msg except Exception as e: error_msg = f"Error copying data: {str(e)}" await ctx.error(error_msg) return error_msg finally: if cursor: cursor.close() if conn: manager.release_connection(conn)