Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
172 changes: 172 additions & 0 deletions examples/_harness.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
from __future__ import annotations

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

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

T = TypeVar("T")


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:]}"


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 _run_recipe_impl(
recipe_call: Callable[[], RecipeOutput | Awaitable[RecipeOutput]],
cleanup: _CleanupTracker,
cleanup_status: ExampleCleanupStatus,
) -> ExampleResult:
"""Shared implementation for running recipes with cleanup."""

async def _run_async() -> RecipeOutput:
try:
result = recipe_call()
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()


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 = ExampleCleanupStatus()
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)
return _run_recipe_impl(lambda: recipe(ctx), cleanup, cleanup_status)

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 = ExampleCleanupStatus()
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)
return _run_recipe_impl(lambda: recipe(ctx, options), cleanup, cleanup_status)

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)
82 changes: 82 additions & 0 deletions examples/devbox_from_blueprint_lifecycle.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
#!/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

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

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

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