insert_row
Add a new row to a CockroachDB table by specifying table name and column values, with optional return of inserted data.
Instructions
Insert a new row into a table.
Args:
table: Table name (schema.table or just table).
data: Column names and values to insert.
returning: Columns to return from inserted row.
Returns:
Insert result.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table | Yes | ||
| data | Yes | ||
| returning | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/cockroachdb_mcp/server.py:256-275 (handler)MCP tool handler for 'insert_row': registered with @mcp.tool() decorator, defines input schema via type hints and docstring, delegates execution to crud.insert_row with error handling.
@mcp.tool() async def insert_row( table: str, data: dict[str, Any], returning: list[str] | None = None, ) -> dict[str, Any]: """Insert a new row into a table. Args: table: Table name (schema.table or just table). data: Column names and values to insert. returning: Columns to return from inserted row. Returns: Insert result. """ try: return await crud.insert_row(table, data, returning) except Exception as e: return {"status": "error", "error": str(e)} - Core helper function implementing the insert_row logic: input validation, dynamic SQL INSERT query construction with optional RETURNING, database execution, and result formatting.
async def insert_row( table: str, data: dict[str, Any], returning: list[str] | None = None, ) -> dict[str, Any]: """Insert a new row into a table. Args: table: Table name (schema.table or just table). data: Column names and values to insert. returning: Columns to return from inserted row. Returns: Insert result. """ # Check read-only mode if settings.read_only: return {"status": "error", "error": "Server is in read-only mode"} # Validate table name valid, error = _validate_table_name(table) if not valid: return {"status": "error", "error": error} if not data: return {"status": "error", "error": "No data provided"} schema, table_name = _parse_table_name(table) # Validate column names for col in data.keys(): if not re.match(r"^[\w]+$", col): return {"status": "error", "error": f"Invalid column name: {col}"} # Build INSERT query columns = list(data.keys()) placeholders = ", ".join(["%s"] * len(columns)) col_list = ", ".join(columns) values = list(data.values()) query = f"INSERT INTO {schema}.{table_name} ({col_list}) VALUES ({placeholders})" # Add RETURNING clause if returning: for col in returning: if not re.match(r"^[\w]+$", col): return {"status": "error", "error": f"Invalid column name in returning: {col}"} query += f" RETURNING {', '.join(returning)}" conn = await connection_manager.ensure_connected() try: async with conn.cursor() as cur: await cur.execute(query, tuple(values)) if returning: row = await cur.fetchone() return { "status": "success", "table": f"{schema}.{table_name}", "action": "inserted", "returning": row, } else: return { "status": "success", "table": f"{schema}.{table_name}", "action": "inserted", "rows_affected": cur.rowcount, } except Exception as e: return {"status": "error", "error": str(e)}