-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathdevbox_tunnel.py
More file actions
117 lines (96 loc) · 3.83 KB
/
devbox_tunnel.py
File metadata and controls
117 lines (96 loc) · 3.83 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
#!/usr/bin/env -S uv run python
"""
---
title: Devbox Tunnel (HTTP Server Access)
slug: devbox-tunnel
use_case: Create a devbox with a tunnel, start an HTTP server, and access the server from the local machine through the tunnel. Uses the async SDK.
workflow:
- Create a devbox with tunnel configuration
- Start an HTTP server inside the devbox
- Make an HTTP request from the local machine through the tunnel
- Validate the response
- Shutdown the devbox
tags:
- devbox
- tunnel
- networking
- http
- async
prerequisites:
- RUNLOOP_API_KEY
run: uv run python -m examples.devbox_tunnel
test: uv run pytest -m smoketest tests/smoketests/examples/
---
"""
from __future__ import annotations
import asyncio
import httpx
from runloop_api_client import AsyncRunloopSDK
from ._harness import run_as_cli, wrap_recipe
from .example_types import ExampleCheck, RecipeOutput, RecipeContext
HTTP_SERVER_PORT = 8080
SERVER_STARTUP_DELAY_S = 2
async def recipe(ctx: RecipeContext) -> RecipeOutput:
"""Create a devbox with a tunnel, start an HTTP server, and access it from the local machine."""
cleanup = ctx.cleanup
sdk = AsyncRunloopSDK()
devbox = await sdk.devbox.create(
name="devbox-tunnel-example",
launch_parameters={
"resource_size_request": "X_SMALL",
},
tunnel={"auth_mode": "open"},
)
cleanup.add(f"devbox:{devbox.id}", devbox.shutdown)
# Start a simple HTTP server inside the devbox using Python's built-in http.server
# We use exec_async because the server runs indefinitely until stopped
server_execution = await devbox.cmd.exec_async(f"python3 -m http.server {HTTP_SERVER_PORT} --directory /tmp")
# Give the server a moment to start
await asyncio.sleep(SERVER_STARTUP_DELAY_S)
# The tunnel was created with the devbox. For authenticated tunnels, set
# tunnel={"auth_mode": "authenticated"} on create and include the auth_token
# in your requests via the Authorization header: `Authorization: Bearer {tunnel.auth_token}`
tunnel = await devbox.get_tunnel()
if tunnel is None:
raise RuntimeError("Failed to create tunnel at devbox launch time")
# Get the tunnel URL for the server port
tunnel_url = await devbox.get_tunnel_url(HTTP_SERVER_PORT)
if tunnel_url is None:
raise RuntimeError("Failed to get tunnel URL after creating tunnel")
# Make an HTTP request from the LOCAL MACHINE through the tunnel to the devbox
# This demonstrates that the tunnel allows external access to the devbox service
async with httpx.AsyncClient() as client:
response = await client.get(tunnel_url)
response_text = response.text
# Stop the HTTP server
await server_execution.kill()
return RecipeOutput(
resources_created=[f"devbox:{devbox.id}"],
checks=[
ExampleCheck(
name="tunnel was created successfully",
passed=bool(tunnel.tunnel_key),
details=f"tunnel_key={tunnel.tunnel_key}",
),
ExampleCheck(
name="tunnel URL was constructed correctly",
passed=bool(
tunnel.tunnel_key and tunnel.tunnel_key in tunnel_url and str(HTTP_SERVER_PORT) in tunnel_url
),
details=tunnel_url,
),
ExampleCheck(
name="HTTP request through tunnel succeeded",
passed=response.is_success,
details=f"status={response.status_code}",
),
ExampleCheck(
name="response contains directory listing",
passed="Directory listing" in response_text,
details=response_text[:200],
),
],
)
run_devbox_tunnel_example = wrap_recipe(recipe)
if __name__ == "__main__":
run_as_cli(run_devbox_tunnel_example)