from beam import endpoint, Image


image = (
    Image(
        base_image="docker.io/nvidia/cuda:12.3.1-runtime-ubuntu20.04",
        python_version="python3.9",
    )
    .add_commands(["apt-get update -y", "apt-get install ffmpeg -y"])
    .add_python_packages(["transformers", "torch"])
    .build_with_gpu(gpu="A10G")
)


@endpoint(image=image)
def handler():
    return {}

from beam import Client, function

@function()
def handler():
    client = Client(
        token="YOUR_TOKEN"
    )

    # Get a deployment by its ID
    deployment = client.get_deployment_by_id("YOUR_DEPLOYMENT_ID")

    # Submit a task
    file = client.upload_file("./app.py")
    task = deployment.submit(input={"audio_file": file})

    task = deployment.submit(input={"text": "Hello world"})

    # Get a deployment by its stub ID
    deployment = client.get_deployment_by_stub_id("YOUR_STUB_ID")
    task = deployment.submit(input={"data": "example"})

    # Get a task by its ID
    task = client.get_task_by_id("YOUR_TASK_ID")
    result = task.result(wait=True)


if __name__ == "__main__":
    handler.remote()

client = Client(token="YOUR_TOKEN")
file_url = client.upload_file("./data.csv")

client = Client(token="YOUR_TOKEN")
task = client.get_task_by_id("YOUR_TASK_ID")
result = task.result(wait=True)

client = Client(token="YOUR_TOKEN")
deployment = client.get_deployment_by_id("YOUR_DEPLOYMENT_ID")
task = deployment.submit(input={"text": "Hello world"})

client = Client(token="YOUR_TOKEN")
deployment = client.get_deployment_by_stub_id("YOUR_STUB_ID")
task = deployment.submit(input={"data": "example"})

# Connect to an existing sandbox
instance = sandbox.connect("sandbox-123")

# Create a new sandbox
instance = sandbox.create()
print(f"Sandbox created with ID: {instance.sandbox_id()}")

# Expose port 8000 for a web service
url = instance.expose_port(8000)
print(f"Web service available at: {url}")

sandbox_id = instance.sandbox_id()
print(f"Working with sandbox: {sandbox_id}")

# Terminate the sandbox
success = instance.terminate()
if success:
    print("Sandbox terminated successfully")

# Keep the sandbox alive for 1 hour
instance.update_ttl(3600)

# Make the sandbox never timeout
instance.update_ttl(-1)

process = pm.exec("sleep", "100")

# Kill the process after 5 seconds
import time
time.sleep(5)
process.kill()

process = pm.exec("python3", "-c", "import sys; print('stdout'); print('stderr', file=sys.stderr)")

# Read combined output
for line in process.logs:
    print(f"LOG: {line.strip()}")

# Or read all at once
all_logs = process.logs.read()

process = pm.exec("sleep", "5")

# Check status periodically
while True:
    exit_code, status = process.status()
    if exit_code >= 0:
        print(f"Process finished with exit code: {exit_code}")
        break
    time.sleep(1)

process = pm.exec("python3", "-c", "import sys; print('Error', file=sys.stderr)")
stderr_content = process.stderr.read()
print(f"STDERR: {stderr_content}")

process = pm.exec("echo", "Hello World")
stdout_content = process.stdout.read()
print(f"STDOUT: {stdout_content}")

process = pm.exec("long_running_command")
exit_code = process.wait()
if exit_code == 0:
    print("Command completed successfully")

# Run a simple command
process = pm.exec("ls", "-la")
process.wait()

# Run with custom environment
process = pm.exec("echo", "$CUSTOM_VAR", env={"CUSTOM_VAR": "hello"})

# Run in specific directory
process = pm.exec("pwd", cwd="/tmp")

try:
    process = pm.get_process(12345)
    print(f"Found process: {process.pid}")
except SandboxProcessError:
    print("Process not found")

processes = pm.list_processes()
for process in processes:
    print(f"Process {process.pid} is running")

# Run blocking Python code
result = pm.run_code("print('Hello from sandbox!')")
print(result.result)

# Run non-blocking Python code
process = pm.run_code("import time; time.sleep(10)", blocking=False)
# Do other work while process runs
process.wait()

# Get a process stream
process = pm.exec("echo", "Hello World")

# Read line by line
for line in process.stdout:
    print(f"Output: {line.strip()}")

# Read all output at once
all_output = process.stdout.read()

# Delete a temporary file
fs.delete_file("/tmp/temp_file.txt")

# Delete a log file
fs.delete_file("/var/log/old_log.log")

# Download a log file
fs.download_file("/var/log/app.log", "local_app.log")

# Download to a specific directory
fs.download_file("/output/result.txt", "./results/result.txt")

# Find all Python files
python_files = fs.find_in_files("/workspace", "*.py")

# Find all log files
log_files = fs.find_in_files("/var/log", "*.log")

# List files in the root directory
files = fs.list_files("/")
for file_info in files:
    if file_info.is_dir:
        print(f"Directory: {file_info.name}")
    else:
        print(f"File: {file_info.name} ({file_info.size} bytes)")

# List files in a specific directory
workspace_files = fs.list_files("/workspace")

# Replace a configuration value
fs.replace_in_files("/config", "old_host", "new_host")

# Update version numbers
fs.replace_in_files("/app", "1.0.0", "1.1.0")

# Get file information
file_info = fs.stat_file("/path/to/file.txt")
print(f"File size: {file_info.size} bytes")
print(f"Is directory: {file_info.is_dir}")
print(f"Modified: {file_info.mod_time}")

# Upload a Python script
fs.upload_file("my_script.py", "/workspace/script.py")

# Upload to a subdirectory
fs.upload_file("config.json", "/app/config/config.json")

from beam import Pod, Image

# Create a Pod that runs a simple HTTP server
pod = Pod(
    name="web-server",
    cpu=2,
    memory="512Mi",
    image=Image(
        base_image="python:3.9-slim",
        python_packages=["requests"],
    ),
    ports=[8000],  # We'll expose port 8000
)

# Spin up the Pod container, running `python -m http.server 8000`
result = pod.create(entrypoint=["python", "-m", "http.server", "8000"])

print("Container ID:", result.container_id)
print("URL:", result.url)

from beam import Pod

pod = Pod(cpu=2, memory="1Gi", ports=[8080])
result = pod.create(entrypoint=["python", "-m", "http.server", "8080"])

if result.ok:
    print("Pod created successfully!")
    print("Container ID:", result.container_id)
    print("URL:", result.url)
else:
    print("Failed to create Pod")

beam deploy app.py:pod

from beam import function


@function(cpu=8)
def square(i: int):
    return i**2


if __name__ == "__main__":
    # Run the `square` function remotely on Beam
    result = square.remote(i=12)
    print(result)

(.venv) user@MacBook % python example.py
=> Building image
=> Using cached image
=> Syncing files
=> Files synced
=> Running function: <example:square>
=> Function complete <908c76b1-ee68-4b33-ac3a-026ae646625f>
144

from beam import function


@function(cpu=8)
def square(i: int):
    return i**2


def main():
    numbers = list(range(10))
    squared = []

    # Run a remote container for every item in list
    for result in square.map(numbers):
        squared.append(result)

from beam import schedule


@schedule(when="*/5 * * * *", name="scheduled-job")
def task():
    print("Hi, from scheduled task!")

from beam import endpoint, Image


@endpoint(
    cpu=1.0,
    memory=128,
    gpu="T4",
    image=Image(python_packages=["torch"]),
    keep_warm_seconds=1000,
)
def multiply(**inputs):
    result = inputs["x"] * 2
    return {"result": result}

beam serve app.py

beam serve app.py

from beam import asgi, Image


@asgi(
    cpu=1.0,
    memory=128,
    gpu="T4",
    image=Image(python_packages=["fastapi"]),
    keep_warm_seconds=10,
    max_pending_tasks=100,
)
def web_server(context):
    from fastapi import FastAPI

    app = FastAPI()

    @app.post("/hello")
    async def something():
        return {"hello": True}

    @app.post("/warmup")
    async def warmup():
        return {"status": "warm"}

    return app

beam serve app.py

from beam import realtime

def generate_text():
    return ["this", "could", "be", "anything"]

@realtime(
    cpu=1.0,
    memory=128,
    gpu="T4"
)
def handler(context):
    return generate_text()

beam serve app.py

from beam import Bot, BotContext, BotLocation, Image
from pydantic import BaseModel


# Define input and output types for the bot
class ProductName(BaseModel):
    product_name: str


class URL(BaseModel):
    url: str


class ReviewPage(BaseModel):
    review_page: str


# Create the bot
bot = Bot(
    model="gpt-4o",
    api_key="YOUR_API_KEY",
    locations=[
        BotLocation(marker=ProductName),
        BotLocation(marker=URL, expose=False),
        BotLocation(marker=ReviewPage, expose=False),
    ],
    description="This bot retrieves product reviews and summarizes them.",
)


# Define a transition
@bot.transition(
    inputs={ProductName: 1},
    outputs=[URL],
    description="Retrieve Google Shopping URLs for a product",
    cpu=1,
    memory=128,
    image=Image(python_packages=["serpapi", "google-search-results"]),
)
def get_product_urls(context: BotContext, inputs):
    product_name = inputs[ProductName][0].product_name
    # Perform some action
    return {URL: [URL(url="https://example.com")]}

from beam import endpoint, QueueDepthAutoscaler


@endpoint(
    autoscaler=QueueDepthAutoscaler(
        min_containers=1, max_containers=3, tasks_per_container=1
    ),
)
def handler():
    return {"success": "true"}

from beam import function, Volume


VOLUME_PATH = "./model_weights"


@function(
    volumes=[Volume(name="model-weights", mount_path=VOLUME_PATH)],
)
def load_model():
    from transformers import AutoModel

    # Load model from cloud storage cache
    AutoModel.from_pretrained(VOLUME_PATH)

from beam import CloudBucket, CloudBucketConfig

# Cloud Bucket
weights = CloudBucket(
    name="weights",
    mount_path="./weights",
    config=CloudBucketConfig(
        access_key="my-access-key",
        secret_key="my-secret-key",
        endpoint="https://s3-endpoint.com",
    ),
)

@function(volumes=[weights])
def my_function():
    pass

from beam import CloudBucketConfig

config = CloudBucketConfig(
    read_only=False,
    access_key="my-access-key",
    secret_key="my-secret-key",
    endpoint="https://s3-endpoint.com",
    region="us-west-2"
)

from beam import Image as BeamImage, Output, function


@function(
    image=BeamImage(
        python_packages=[
            "pillow",
        ],
    ),
)
def save_image():
    from PIL import Image as PILImage

    # Generate PIL image
    pil_image = PILImage.new(
        "RGB", (100, 100), color="white"
    )  # Creating a 100x100 white image

    # Save image file
    output = Output.from_pil_image(pil_image)
    output.save()

    # Retrieve pre-signed URL for output file
    url = output.public_url(expires=400)
    print(url)

    # Print other details about the output
    print(f"Output ID: {output.id}")
    print(f"Output Path: {output.path}")
    print(f"Output Stats: {output.stat()}")
    print(f"Output Exists: {output.exists()}")

    return {"image": url}


if __name__ == "__main__":
    save_image()

https://app.stage.beam.cloud/output/id/abe0c95a-2cd1-40b3-bace-9225f2c79c6d