SQL Server MCP

Instructions

Args:
    table: Table name (can include schema: 'dbo.Users' or 'Users')
    id: Primary key value of the row to update
    data: Dictionary of column names and new values

Returns:
    Dictionary with:
    - status: 'success' or error
    - table: Full table name
    - updated: The updated row

Implementation Reference

• src/mssql_mcp/tools/crud.py:191-254 (handler)

  The core handler function for the 'update_row' tool, decorated with @mcp.tool() which registers it in the MCP system. It performs an SQL UPDATE on the specified table row by primary key, using OUTPUT to return the updated row.

  @mcp.tool()
  def update_row(table: str, id: Any, data: dict[str, Any]) -> dict[str, Any]:
      """Update an existing row by primary key.
  
      Args:
          table: Table name (can include schema: 'dbo.Users' or 'Users')
          id: Primary key value of the row to update
          data: Dictionary of column names and new values
  
      Returns:
          Dictionary with:
          - status: 'success' or error
          - table: Full table name
          - updated: The updated row
      """
      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 update"}
  
          # Get primary key column
          pk_cols = _get_primary_key_columns(schema, table_name)
          if not pk_cols:
              return {"error": f"No primary key found for table {schema}.{table_name}"}
  
          # Build UPDATE statement with OUTPUT clause
          set_clauses = ", ".join([f"[{k}] = %s" for k in data])
          params = tuple(data.values()) + (id,)
  
          query = f"""
              UPDATE [{schema}].[{table_name}]
              SET {set_clauses}
              OUTPUT INSERTED.*
              WHERE [{pk_cols[0]}] = %s
          """
  
          rows = manager.execute_query(query, params)
  
          if not rows:
              return {
                  "error": f"No row found with {pk_cols[0]} = {id}",
                  "table": f"{schema}.{table_name}",
              }
  
          return {
              "status": "success",
              "table": f"{schema}.{table_name}",
              "updated": rows[0],
          }
  
      except QueryError as e:
          logger.error(f"Error updating row in {table}: {e}")
          return {"error": str(e)}
      except Exception as e:
          logger.error(f"Unexpected error updating row in {table}: {e}")
          return {"error": str(e)}