SQL Server MCP

Instructions

Args:
    table: Table name (can include schema: 'dbo.Users' or 'Users')
    data: Dictionary of column names and values to insert

Returns:
    Dictionary with:
    - status: 'success' or error
    - table: Full table name
    - inserted: The inserted row (including generated identity columns)

Implementation Reference

• src/mssql_mcp/tools/crud.py:134-189 (handler)

  The `insert_row` tool handler function, decorated with `@mcp.tool()` for registration in the MCP server. Handles inserting a new row into a specified table using a parameterized INSERT query with OUTPUT clause to return the inserted data, including auto-generated columns. Includes safety checks for read-only mode and empty data.

  @mcp.tool()
  def insert_row(table: str, data: dict[str, Any]) -> dict[str, Any]:
      """Insert a new row into a table.
  
      Args:
          table: Table name (can include schema: 'dbo.Users' or 'Users')
          data: Dictionary of column names and values to insert
  
      Returns:
          Dictionary with:
          - status: 'success' or error
          - table: Full table name
          - inserted: The inserted row (including generated identity columns)
      """
      try:
          manager = get_connection_manager()
          config = manager.config
  
          # Check read-only mode
          if config.read_only:
              return {"error": "Write operations disabled in read-only mode"}
  
          schema, table_name = parse_table_name(table)
  
          if not data:
              return {"error": "No data provided for insert"}
  
          # Build INSERT statement with OUTPUT clause
          cols = ", ".join([f"[{k}]" for k in data])
          placeholders = ", ".join(["%s"] * len(data))
          values = tuple(data.values())
  
          query = f"""
              INSERT INTO [{schema}].[{table_name}] ({cols})
              OUTPUT INSERTED.*
              VALUES ({placeholders})
          """
  
          rows = manager.execute_query(query, values)
  
          # The OUTPUT clause returns the inserted row
          inserted = rows[0] if rows else data
  
          return {
              "status": "success",
              "table": f"{schema}.{table_name}",
              "inserted": inserted,
          }
  
      except QueryError as e:
          logger.error(f"Error inserting row into {table}: {e}")
          return {"error": str(e)}
      except Exception as e:
          logger.error(f"Unexpected error inserting row into {table}: {e}")
          return {"error": str(e)}