File Operations

from skyvern import Skyvern

skyvern = Skyvern(api_key="your-api-key")

browser = await skyvern.launch_cloud_browser()
page = await browser.get_working_page()

await page.goto("https://example.com/upload")

# Upload using natural language prompt
await page.upload_file(
    prompt="Upload resume file",
    files="/path/to/resume.pdf"
)

# Or use traditional selector with AI fallback
await page.upload_file(
    "#file-input",
    prompt="Upload document",
    files="/path/to/document.docx"
)

# Upload multiple files at once
await page.upload_file(
    prompt="Attach supporting documents",
    files=[
        "/path/to/resume.pdf",
        "/path/to/cover_letter.pdf",
        "/path/to/references.pdf"
    ]
)

task = await skyvern.run_task(
    prompt="Go to the job application form and upload the resume file located at resume.pdf",
    url="https://careers.example.com/apply",
    # Files are automatically handled by Skyvern
)

from skyvern.webeye.actions.actions import UploadFileAction

# Skyvern creates UploadFileAction when it detects file inputs
action = UploadFileAction(
    file_url="s3://bucket/path/to/file.pdf",  # S3 URI
    # OR
    file_name="document.pdf",  # Local file name
    is_upload_file_tag=True,  # Detected as file input
)

task = await skyvern.run_task(
    prompt="Download the latest invoice PDF",
    url="https://example.com/invoices"
)

# Access downloaded files
for file in task.downloaded_files:
    print(f"Downloaded: {file.file_name}")
    print(f"URL: {file.presigned_url}")  # Presigned URL to download
    print(f"Size: {file.file_size} bytes")

# Downloads are saved to: /app/downloads/{org_id}/{browser_session_id}/
# Files are automatically uploaded to block storage if configured

from skyvern.webeye.actions.actions import DownloadFileAction

# Skyvern creates DownloadFileAction for download links
action = DownloadFileAction(
    file_url="https://example.com/files/document.pdf",
    download=True,  # Trigger download
    download_triggered=True,  # Download was initiated
    downloaded_files=["invoice_2024.pdf"],  # Files downloaded
)

# Automatic download timeout: 30 seconds default
# Max wait time for download completion: 60 seconds

# Downloads are monitored via Chrome DevTools Protocol (CDP)
await page.click(prompt="Download report")
# Skyvern automatically waits for download to complete

# In a workflow, use FileParserBlock
from skyvern.forge.sdk.workflow.models.block import BlockType

# Workflow YAML example:
workflow = {
    "blocks": [
        {
            "block_type": "file_url_parser",
            "label": "parse_invoice",
            "file_url": "s3://bucket/invoices/latest.pdf",
            "file_type": "pdf"
        },
        {
            "block_type": "task",
            "label": "extract_data",
            "prompt": "Extract invoice number and total from: {{ parse_invoice.output }}"
        }
    ]
}

blocks:
  - block_type: file_url_parser
    label: parse_document
    file_url: "{{ previous_block.downloaded_files[0].s3_uri }}"
    file_type: pdf
    
  - block_type: extraction
    label: extract_invoice_data
    data_extraction_goal: |
      Extract the following from the parsed document:
      - Invoice number
      - Invoice date
      - Total amount
      - Vendor name
    data_schema:
      type: object
      properties:
        invoice_number:
          type: string
        invoice_date:
          type: string
        total_amount:
          type: number
        vendor_name:
          type: string
    context: "{{ parse_document.output }}"

# Via API
response = await skyvern.upload_file(
    file_path="/path/to/document.pdf",
    file_name="document.pdf"
)

print(f"S3 URI: {response.s3_uri}")
print(f"Download URL: {response.presigned_url}")

curl -X POST https://api.skyvern.com/api/v1/files/upload \
  -H "x-api-key: your-api-key" \
  -F "file=@/path/to/document.pdf" \
  -F "file_name=document.pdf"

{
  "s3_uri": "s3://skyvern-uploads/org_123/document.pdf",
  "presigned_url": "https://s3.amazonaws.com/..."
}

from skyvern.forge.sdk.schemas.files import FileInfo

class FileInfo(BaseModel):
    file_name: str              # Original file name
    file_size: int              # Size in bytes
    s3_uri: str                 # S3 storage location
    presigned_url: str          # Temporary download URL
    content_type: str | None    # MIME type

workflow_run = await skyvern.run_workflow(
    workflow_id="wpid_invoice_processor",
    parameters={
        "vendor_portal_url": "https://vendor.example.com"
    }
)

# Workflow steps:
# 1. Login to vendor portal
# 2. Navigate to invoices
# 3. Download all invoices from last month
# 4. Parse each PDF
# 5. Extract invoice data
# 6. Upload to accounting system

for file in workflow_run.downloaded_files:
    print(f"Processed: {file.file_name}")

import asyncio
from pathlib import Path

async def upload_documents(directory: str):
    """Upload all PDFs from a directory"""
    files = list(Path(directory).glob("*.pdf"))
    
    # Upload to storage first
    uploaded_files = []
    for file_path in files:
        response = await skyvern.upload_file(
            file_path=str(file_path),
            file_name=file_path.name
        )
        uploaded_files.append(response.s3_uri)
    
    # Use in workflow
    await skyvern.run_workflow(
        workflow_id="wpid_document_processor",
        parameters={
            "document_uris": uploaded_files
        }
    )

blocks:
  # Step 1: Download file
  - block_type: task
    label: download_report
    url: https://example.com/reports
    navigation_goal: Download the latest quarterly report
    
  # Step 2: Parse downloaded file
  - block_type: file_url_parser
    label: parse_report
    file_url: "{{ download_report.downloaded_files[0].s3_uri }}"
    file_type: pdf
    
  # Step 3: Extract data from parsed content
  - block_type: extraction
    label: extract_metrics
    data_extraction_goal: Extract quarterly revenue and profit
    context: "{{ parse_report.output }}"
    data_schema:
      type: object
      properties:
        revenue:
          type: number
        profit:
          type: number
        quarter:
          type: string

# Parse source document
response = await skyvern.upload_file(
    file_path="employee_data.csv",
    file_name="employee_data.csv"
)

# Use parsed data in form filling
task = await skyvern.run_task(
    prompt=f"""
    Fill out the employee registration form using data from the CSV.
    For each row in the CSV, create a new employee entry.
    """,
    url="https://hr.example.com/register",
    data_extraction_schema={
        "employees_processed": "number",
        "errors": "array"
    }
)

# Good
await page.upload_file(
    prompt="Upload company logo in the profile section",
    files="logo.png"
)

# Less specific
await page.upload_file(
    prompt="Upload file",
    files="logo.png"
)

from pathlib import Path

file_path = Path("/path/to/large_file.pdf")
if file_path.stat().st_size > 10 * 1024 * 1024:  # 10MB
    print("File too large, consider splitting")

try:
    await page.upload_file(
        prompt="Upload document",
        files="document.pdf"
    )
except Exception as e:
    print(f"Upload failed: {e}")
    # Retry or handle error

task = await skyvern.run_task(
    prompt="Download invoice",
    url="https://example.com/invoices"
)

if not task.downloaded_files:
    print("No files were downloaded")
    # Handle missing download
else:
    for file in task.downloaded_files:
        if file.file_size == 0:
            print(f"Warning: {file.file_name} is empty")

# Presigned URLs expire after a certain time
for file in task.downloaded_files:
    # Download immediately or save URL for short-term use
    async with httpx.AsyncClient() as client:
        response = await client.get(file.presigned_url)
        with open(f"local_{file.file_name}", "wb") as f:
            f.write(response.content)

- block_type: file_url_parser
  label: parse_doc
  file_url: "{{ file_uri }}"
  
- block_type: validation
  label: check_content
  validation_goal: Verify the parsed document contains expected sections

# For large documents, consider pagination or chunking
# Parse in smaller sections if needed

# Ensure file path is absolute
import os
absolute_path = os.path.abspath("relative/path/to/file.pdf")

await page.upload_file(
    prompt="Upload file",
    files=absolute_path
)

# Be more specific in prompt
await page.upload_file(
    prompt="Upload file to the main document field, not the thumbnail field",
    files="document.pdf"
)

# Increase timeout for large files
# Configure via environment or task settings