Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/variable.py:111-157 (handler)

  The @mcp.tool decorated async function implementing the 'update_variable' MCP tool. It updates a Prefect variable by name, optionally changing its value and tags, using the Prefect client. Handles existence check, updates, and returns confirmation with updated details.

  @mcp.tool
  async def update_variable(
      name: str,
      value: Optional[Any] = None,
      tags: Optional[List[str]] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Update a variable.
      
      Args:
          name: The variable name
          value: New value (can be any JSON-serializable type)
          tags: New tags
          
      Returns:
          Confirmation message or error
      """
      try:
          async with get_client() as client:
              from prefect.client.schemas.actions import VariableUpdate
              
              # Check if variable exists first
              existing_variable = await client.read_variable_by_name(name)
              if existing_variable is None:
                  return [types.TextContent(type="text", text=json.dumps({"error": f"Variable '{name}' not found"}, indent=2))]
              
              # Prepare update data
              update_data = {}
              if value is not None:
                  update_data["value"] = value
              if tags is not None:
                  update_data["tags"] = tags
              
              # Create update object
              variable_update = VariableUpdate(name=name, **update_data)
              
              await client.update_variable(variable=variable_update)
              
              # Read the updated variable to return its details
              updated_variable = await client.read_variable_by_name(name)
              
              return [types.TextContent(type="text", text=json.dumps({
                  "message": f"Variable '{name}' updated successfully",
                  "variable": updated_variable.model_dump()
              }, indent=2, default=str))]
      except Exception as e:
          return [types.TextContent(type="text", text=json.dumps({"error": str(e)}, indent=2))]