Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@ jobs:
- name: Run lints
run: ./scripts/lint

- name: Verify generated examples docs
run: uv run python scripts/generate_examples_md.py --check

build:
if: github.event_name == 'push' || github.event.pull_request.head.repo.fork
timeout-minutes: 10
Expand Down
4 changes: 4 additions & 0 deletions .github/workflows/smoketests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,10 @@ jobs:
fi
echo "DEBUG=false" >> $GITHUB_ENV
echo "PYTHONPATH=${{ github.workspace }}/src" >> $GITHUB_ENV
echo "RUN_EXAMPLE_LIVE_TESTS=1" >> $GITHUB_ENV

- name: Verify generated example artifacts
run: uv run python scripts/generate_examples_md.py --check

- name: Run smoke tests (pytest via uv)
env:
Expand Down
72 changes: 72 additions & 0 deletions EXAMPLES.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
# Examples

> This file is auto-generated from metadata in `examples/*.py`.
> Do not edit this file manually. Run `uv run python scripts/generate_examples_md.py` instead.

Runnable examples live in [`examples/`](./examples).

## Table of Contents

- [Devbox From Blueprint (Run Command, Shutdown)](#devbox-from-blueprint-lifecycle)
- [MCP Hub + Claude Code + GitHub](#mcp-github-tools)

<a id="devbox-from-blueprint-lifecycle"></a>
## Devbox From Blueprint (Run Command, Shutdown)

**Use case:** Create a devbox from a blueprint, run a command, validate output, and cleanly tear everything down.

**Tags:** `devbox`, `blueprint`, `commands`, `cleanup`

### Workflow
- Create a blueprint
- Create a devbox from the blueprint
- Execute a command in the devbox
- Validate exit code and stdout
- Shutdown devbox and delete blueprint

### Prerequisites
- `RUNLOOP_API_KEY`

### Run
```sh
uv run python -m examples.devbox_from_blueprint_lifecycle
```

### Test
```sh
uv run pytest -m smoketest tests/smoketests/examples/
```

**Source:** [`examples/devbox_from_blueprint_lifecycle.py`](./examples/devbox_from_blueprint_lifecycle.py)

<a id="mcp-github-tools"></a>
## MCP Hub + Claude Code + GitHub

**Use case:** Connect Claude Code running in a devbox to GitHub tools through MCP Hub without exposing raw GitHub credentials to the devbox.

**Tags:** `mcp`, `devbox`, `github`, `commands`, `cleanup`

### Workflow
- Create an MCP config for GitHub
- Store GitHub token as a Runloop secret
- Launch a devbox with MCP Hub wiring
- Install Claude Code and register MCP endpoint
- Run a Claude prompt through MCP tools
- Shutdown devbox and clean up cloud resources

### Prerequisites
- `RUNLOOP_API_KEY`
- `GITHUB_TOKEN (GitHub PAT with repo scope)`
- `ANTHROPIC_API_KEY`

### Run
```sh
GITHUB_TOKEN=ghp_xxx ANTHROPIC_API_KEY=sk-ant-xxx uv run python -m examples.mcp_github_tools
```

### Test
```sh
uv run pytest -m smoketest tests/smoketests/examples/
```

**Source:** [`examples/mcp_github_tools.py`](./examples/mcp_github_tools.py)
17 changes: 17 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,23 @@ asyncio.run(main())

Functionality between the synchronous and asynchronous clients is otherwise identical.

## Examples

Workflow-oriented runnable examples are documented in [`EXAMPLES.md`](./EXAMPLES.md).

`EXAMPLES.md` is generated from metadata in `examples/*.py` and should not be edited manually.
Regenerate it with:

```sh
uv run python scripts/generate_examples_md.py
```

## Agent Guidance

Detailed agent-specific instructions for developing using this package live in [`llms.txt`](./llms.txt). Consolidated recipes for frequent tasks are in [`EXAMPLES.md`](./EXAMPLES.md).

After completing any modifications to this project, ensure `llms.txt` and `README.md` are kept in sync.

### With aiohttp

By default, the async client uses `httpx` for HTTP requests. However, for improved concurrency performance you may also use `aiohttp` as the HTTP backend.
Expand Down
4 changes: 0 additions & 4 deletions examples/.keep

This file was deleted.

1 change: 1 addition & 0 deletions examples/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# Examples package for Runloop Python SDK
178 changes: 178 additions & 0 deletions examples/_harness.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,178 @@
from __future__ import annotations

import sys
import json
import asyncio
from typing import Any, TypeVar, Callable, Awaitable
from dataclasses import asdict

from .example_types import (
ExampleCheck,
RecipeOutput,
ExampleResult,
RecipeContext,
ExampleCleanupStatus,
ExampleCleanupFailure,
empty_cleanup_status,
)

T = TypeVar("T")


class _CleanupTracker:
"""Tracks cleanup actions and executes them in LIFO order."""

def __init__(self, status: ExampleCleanupStatus) -> None:
self._status = status
self._actions: list[tuple[str, Callable[[], Any]]] = []

def add(self, resource: str, action: Callable[[], Any]) -> None:
"""Register a cleanup action for a resource."""
self._actions.append((resource, action))

async def run(self) -> None:
"""Execute all cleanup actions in reverse order."""
while self._actions:
resource, action = self._actions.pop()
self._status.attempted.append(resource)
try:
result = action()
if asyncio.iscoroutine(result):
await result
self._status.succeeded.append(resource)
except Exception as e:
self._status.failed.append(ExampleCleanupFailure(resource, str(e)))

if self._status.attempted:
if not self._status.failed:
print("Cleanup completed.") # noqa: T201
else:
print("Cleanup finished with errors.") # noqa: T201


def _should_fail_process(result: ExampleResult) -> bool:
"""Determine if the process should exit with failure."""
has_failed_checks = any(not check.passed for check in result.checks)
return result.skipped or has_failed_checks or len(result.cleanup_status.failed) > 0


def wrap_recipe(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

basically a duplicate of wrap_recipe_with_options right? could we clean up/refactor?

recipe: Callable[[RecipeContext], RecipeOutput] | Callable[[RecipeContext], Awaitable[RecipeOutput]],
validate_env: Callable[[], tuple[bool, list[ExampleCheck]]] | None = None,
) -> Callable[[], ExampleResult]:
"""Wrap a recipe function with cleanup tracking and result handling.

Args:
recipe: The recipe function to wrap. Can be sync or async.
validate_env: Optional function to validate environment before running.
Returns (skip, checks) tuple.

Returns:
A callable that runs the recipe and returns ExampleResult.
"""

def run() -> ExampleResult:
cleanup_status = empty_cleanup_status()
cleanup = _CleanupTracker(cleanup_status)

if validate_env is not None:
skip, checks = validate_env()
if skip:
return ExampleResult(
resources_created=[],
checks=checks,
cleanup_status=cleanup_status,
skipped=True,
)

ctx = RecipeContext(cleanup=cleanup)

async def _run_async() -> RecipeOutput:
try:
result = recipe(ctx)
if asyncio.iscoroutine(result):
output: RecipeOutput = await result
return output
return result # type: ignore[return-value]
finally:
await cleanup.run()

loop = asyncio.new_event_loop()
try:
output = loop.run_until_complete(_run_async())
return ExampleResult(
resources_created=output.resources_created,
checks=output.checks,
cleanup_status=cleanup_status,
)
finally:
loop.close()

return run


def wrap_recipe_with_options(
recipe: Callable[[RecipeContext, T], RecipeOutput] | Callable[[RecipeContext, T], Awaitable[RecipeOutput]],
validate_env: Callable[[T], tuple[bool, list[ExampleCheck]]] | None = None,
) -> Callable[[T], ExampleResult]:
"""Wrap a recipe function that takes options with cleanup tracking.

Args:
recipe: The recipe function to wrap. Can be sync or async. Takes options parameter.
validate_env: Optional function to validate environment before running.
Takes options and returns (skip, checks) tuple.

Returns:
A callable that runs the recipe with options and returns ExampleResult.
"""

def run(options: T) -> ExampleResult:
cleanup_status = empty_cleanup_status()
cleanup = _CleanupTracker(cleanup_status)

if validate_env is not None:
skip, checks = validate_env(options)
if skip:
return ExampleResult(
resources_created=[],
checks=checks,
cleanup_status=cleanup_status,
skipped=True,
)

ctx = RecipeContext(cleanup=cleanup)

async def _run_async() -> RecipeOutput:
try:
result = recipe(ctx, options)
if asyncio.iscoroutine(result):
output: RecipeOutput = await result
return output
return result # type: ignore[return-value]
finally:
await cleanup.run()

loop = asyncio.new_event_loop()
try:
output = loop.run_until_complete(_run_async())
return ExampleResult(
resources_created=output.resources_created,
checks=output.checks,
cleanup_status=cleanup_status,
)
finally:
loop.close()

return run


def run_as_cli(run: Callable[[], ExampleResult]) -> None:
"""Run an example and exit with appropriate status code."""
try:
result = run()
print(json.dumps(asdict(result), indent=2)) # noqa: T201
if _should_fail_process(result):
sys.exit(1)
except Exception as e:
print(f"Error: {e}") # noqa: T201
sys.exit(1)
90 changes: 90 additions & 0 deletions examples/devbox_from_blueprint_lifecycle.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
#!/usr/bin/env -S uv run python
"""
---
title: Devbox From Blueprint (Run Command, Shutdown)
slug: devbox-from-blueprint-lifecycle
use_case: Create a devbox from a blueprint, run a command, validate output, and cleanly tear everything down.
workflow:
- Create a blueprint
- Create a devbox from the blueprint
- Execute a command in the devbox
- Validate exit code and stdout
- Shutdown devbox and delete blueprint
tags:
- devbox
- blueprint
- commands
- cleanup
prerequisites:
- RUNLOOP_API_KEY
run: uv run python -m examples.devbox_from_blueprint_lifecycle
test: uv run pytest -m smoketest tests/smoketests/examples/
---
"""

from __future__ import annotations

import time

from runloop_api_client import RunloopSDK
from runloop_api_client.lib.polling import PollingConfig

from ._harness import run_as_cli, wrap_recipe
from .example_types import ExampleCheck, RecipeOutput, RecipeContext


def unique_name(prefix: str) -> str:
"""Generate a unique name with timestamp and random suffix."""
return f"{prefix}-{int(time.time())}-{hex(int(time.time() * 1000) % 0xFFFFFF)[2:]}"


BLUEPRINT_POLL_TIMEOUT_S = 10 * 60


def recipe(ctx: RecipeContext) -> RecipeOutput:
"""Create a devbox from a blueprint, run a command, and clean up."""
cleanup = ctx.cleanup

sdk = RunloopSDK()

blueprint = sdk.blueprint.create(
name=unique_name("example-blueprint"),
dockerfile='FROM ubuntu:22.04\nRUN echo "Hello from your blueprint"',
polling_config=PollingConfig(timeout_seconds=BLUEPRINT_POLL_TIMEOUT_S),
)
cleanup.add(f"blueprint:{blueprint.id}", blueprint.delete)

devbox = blueprint.create_devbox(
name=unique_name("example-devbox"),
launch_parameters={
"resource_size_request": "X_SMALL",
"keep_alive_time_seconds": 60 * 5,
},
)
cleanup.add(f"devbox:{devbox.id}", devbox.shutdown)

result = devbox.cmd.exec('echo "Hello from your devbox"')
stdout = result.stdout()

return RecipeOutput(
resources_created=[f"blueprint:{blueprint.id}", f"devbox:{devbox.id}"],
checks=[
ExampleCheck(
name="command exits successfully",
passed=result.exit_code == 0,
details=f"exitCode={result.exit_code}",
),
ExampleCheck(
name="command output contains expected text",
passed="Hello from your devbox" in stdout,
details=stdout.strip(),
),
],
)


run_devbox_from_blueprint_lifecycle_example = wrap_recipe(recipe)


if __name__ == "__main__":
run_as_cli(run_devbox_from_blueprint_lifecycle_example)
Loading