update_data
Update records in a MySQL table by specifying the table, field-value pairs for new data, and a WHERE condition with parameters. Supports optional database configuration.
Instructions
更新表中的数据
Args:
table_name: 表名
data: 要更新的数据，字段名和值的字典
condition: WHERE条件子句
params: 条件参数列表
db_config: 数据库连接配置参数，如果为None则使用默认配置
Returns:
包含更新结果的字典Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table_name | Yes | ||
| data | Yes | ||
| condition | Yes | ||
| params | No | ||
| db_config | No |
Implementation Reference
- mysql-mcp.py:351-407 (handler)The 'update_data' tool handler function decorated with @mcp.tool(). Updates rows in a MySQL table by building a SET clause from the data dict and applying a WHERE condition with parameterized queries.
@mcp.tool() async def update_data(table_name: str, data: Dict[str, Any], condition: str, params: Optional[List[Any]] = None, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """更新表中的数据 Args: table_name: 表名 data: 要更新的数据，字段名和值的字典 condition: WHERE条件子句 params: 条件参数列表 db_config: 数据库连接配置参数，如果为None则使用默认配置 Returns: 包含更新结果的字典 """ if not table_name or not data or not condition: return {"error": "表名、数据和条件不能为空"} if params is None: params = [] try: conn = get_connection(db_config) cursor = conn.cursor() set_clause = ", ".join([f"{key} = %s" for key in data.keys()]) values = list(data.values()) + params update_sql = f"UPDATE {table_name} SET {set_clause} WHERE {condition}" cursor.execute(update_sql, values) conn.commit() return { "success": True, "affected_rows": cursor.rowcount, "message": f"表 {table_name} 中的数据更新成功" } except Error as e: error_message = f"更新数据失败: {str(e)}" if "doesn't exist" in str(e): error_message += f"\n原因：表 {table_name} 不存在" elif "Unknown column" in str(e): error_message += "\n原因：更新数据中包含表中不存在的列" elif "cannot be null" in str(e): error_message += "\n原因：某个NOT NULL字段被设置为NULL值" elif "Duplicate entry" in str(e): error_message += "\n原因：更新后的数据违反了唯一键约束" elif "Data too long" in str(e): error_message += "\n原因：更新的数据超出了字段的长度限制" elif "syntax error" in str(e).lower(): error_message += f"\n原因：WHERE条件 '{condition}' 存在语法错误" return {"error": error_message} except Exception as e: return {"error": f"更新数据时发生未知错误: {str(e)}"} finally: if 'conn' in locals() and conn.is_connected(): cursor.close() conn.close() - mysql-mcp.py:351-351 (registration)The '@mcp.tool()' decorator that registers 'update_data' as an MCP tool.
@mcp.tool()