add examples and lib

Author: akosyakov

.devcontainer/Dockerfile

@@ -1,9 +1,8 @@
-ARG VARIANT="3.9"
-FROM mcr.microsoft.com/vscode/devcontainers/python:0-${VARIANT}
+FROM mcr.microsoft.com/devcontainers/python:3.12
 
 USER vscode
 
 RUN curl -sSf https://rye.astral.sh/get | RYE_VERSION="0.35.0" RYE_INSTALL_OPTION="--yes" bash
 ENV PATH=/home/vscode/.rye/shims:$PATH
 
-RUN echo "[[ -d .venv ]] && source .venv/bin/activate" >> /home/vscode/.bashrc
+RUN echo "[[ -d .venv ]] && source .venv/bin/activate || export PATH=\$PATH" >> /home/vscode/.bashrc

.devcontainer/devcontainer.json

@@ -24,6 +24,9 @@
         }
       }
     }
+  },
+  "features": {
+    "ghcr.io/devcontainers/features/node:1": {}
   }
 
   // Features to add to the dev container. More info: https://containers.dev/features.

examples/anthropic_tool_use.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python
+
+from __future__ import annotations
+
+from typing import cast
+
+from anthropic import Anthropic
+from anthropic.types import ToolParam, MessageParam
+
+import gitpod.lib as util
+from gitpod import AsyncGitpod
+from gitpod.types.environment_initializer_param import Spec
+
+gpclient = AsyncGitpod()
+llmclient = Anthropic()
+
+user_message: MessageParam = {
+    "role": "user",
+    "content": "What is the test coverage for this repository: https://github.com/gitpod-io/gitpod-sdk-go",
+}
+tools: list[ToolParam] = [
+    {
+        "name": "create_environment",
+        "description": "Create a new environment for a given context URL. This will create a new environment and return the ID of the environment.",
+        "input_schema": {
+            "type": "object",
+            "properties": {"context_url": {"type": "string"}},
+        },
+    },
+    {
+        "name": "execute_command",
+        "description": "Execute a command in a given environment ID. This will execute the command in the given environment and return the output of the command.",
+        "input_schema": {
+            "type": "object",
+            "properties": {"environment_id": {"type": "string"}, "command": {"type": "string"}},
+        },
+    },
+]
+
+async def create_environment(args: dict[str, str], cleanup: util.Disposables) -> str:
+    env_class = await util.find_most_used_environment_class(gpclient)
+    if not env_class:
+        raise Exception("No environment class found. Please create one first.")
+    env_class_id = env_class.id
+    assert env_class_id is not None
+
+    environment = (await gpclient.environments.create(
+        spec={
+            "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
+            "content": {
+                "initializer": {"specs": [Spec(
+                    context_url={
+                        "url": args["context_url"]
+                    }
+                )]},
+            },
+            "machine": {"class": env_class_id},
+        }
+    )).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
+    cleanup.add(lambda: asyncio.run(gpclient.environments.delete(environment_id=environment_id)))    
+    
+    print(f"\nCreated environment: {environment_id} - waiting for it to be ready...")
+    await util.wait_for_environment_ready(gpclient, environment_id)
+    print(f"\nEnvironment is ready: {environment_id}")
+    return environment_id
+
+async def execute_command(args: dict[str, str]) -> str:
+    lines_iter = await util.run_command(gpclient, args["environment_id"], args["command"])
+    lines: list[str] = []
+    async for line in lines_iter:
+        lines.append(line)
+    return "\n".join(lines)
+
+async def main(cleanup: util.Disposables) -> None:
+    messages = [user_message]
+    while True:
+        message = llmclient.messages.create(
+            model="claude-3-5-sonnet-latest",
+            max_tokens=1024,
+            messages=messages,
+            tools=tools,
+        )
+        print(f"\nResponse: {message.model_dump_json(indent=2)}")
+        
+        if message.stop_reason != "tool_use":
+            print(f"\nFinal response reached! {message.model_dump_json(indent=2)}")
+            break
+        
+        messages.extend([
+            {"role": message.role, "content": message.content}
+        ])
+        
+        # Handle all tool calls in this response
+        for tool in (c for c in message.content if c.type == "tool_use"):
+            try:
+                if tool.name == "create_environment":
+                    args = cast(dict[str, str], tool.input)
+                    environment_id = await create_environment(args, cleanup)
+                    messages.append({
+                        "role": "user",
+                        "content": [{
+                                "type": "tool_result",
+                                "tool_use_id": tool.id,
+                                "content": [{"type": "text", "text": f"The environment ID is {environment_id}"}],
+                        }],
+                    })
+                elif tool.name == "execute_command":
+                    args = cast(dict[str, str], tool.input)
+                    output = await execute_command(args)
+                    messages.append({
+                        "role": "user",
+                        "content": [{
+                            "type": "tool_result",
+                            "tool_use_id": tool.id,
+                            "content": [{"type": "text", "text": output}],
+                        }],
+                    })
+                else:
+                    raise Exception(f"Unknown tool: {tool.name}")
+            except Exception as e:
+                messages.append({
+                    "role": "user",
+                    "content": [{
+                        "type": "tool_result",
+                        "tool_use_id": tool.id,
+                        "is_error": True,
+                        "content": [{"type": "text", "text": f"Error: {e}"}],
+                    }],
+                })
+
+    print("\nFinal response reached!")
+
+if __name__ == "__main__":
+    import asyncio
+    disposables = util.Disposables()
+    with disposables:
+        asyncio.run(main(disposables))

examples/fs_access.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python
+
+import sys
+import asyncio
+from io import StringIO
+
+import paramiko
+
+import gitpod.lib as util
+from gitpod import AsyncGitpod
+from gitpod.types.environment_spec_param import EnvironmentSpecParam
+from gitpod.types.environment_initializer_param import Spec
+
+
+# Examples:
+# - ./examples/fs_access.py
+# - ./examples/fs_access.py https://github.com/gitpod-io/empty
+async def main(cleanup: util.Disposables) -> None:
+    client = AsyncGitpod()
+
+    context_url = sys.argv[1] if len(sys.argv) > 1 else None
+
+    env_class = await util.find_most_used_environment_class(client)
+    if not env_class:
+        print("Error: No environment class found. Please create one first.")
+        sys.exit(1)
+    print(f"Found environment class: {env_class.display_name} ({env_class.description})")
+    env_class_id = env_class.id
+    assert env_class_id is not None
+    
+    print("Generating SSH key pair")
+    key = paramiko.RSAKey.generate(2048)
+    private_key_file = StringIO()
+    key.write_private_key(private_key_file)
+    private_key_file.seek(0)  # Reset position to start
+    public_key = f"{key.get_name()} {key.get_base64()}"
+
+    print("Creating environment with SSH access")
+    key_id = "fs-access-example"
+    spec: EnvironmentSpecParam = {
+        "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
+        "machine": {"class": env_class_id},
+        "ssh_public_keys": [{
+            "id": key_id,
+            "value": public_key
+        }]
+    }
+    if context_url:
+        spec["content"] = {
+            "initializer": {"specs": [Spec(
+                context_url={
+                    "url": context_url
+                }
+            )]}
+        }
+
+    print("Creating environment")
+    environment = (await client.environments.create(spec=spec)).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
+    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
+
+    env = util.EnvironmentState(client, environment_id)
+    cleanup.add(lambda: asyncio.run(env.close()))
+
+    print("Waiting for environment to be running")
+    await env.wait_until_running()
+
+    print("Waiting for SSH key to be applied")
+    await env.wait_for_ssh_key_applied(key_id=key_id, key_value=public_key)
+
+    print("Waiting for SSH URL")
+    ssh_url = await env.wait_for_ssh_url()
+
+    print(f"Setting up SSH connection to {ssh_url}")
+    # Parse ssh://username@host:port format
+    url_parts = ssh_url.split('://')[-1]
+    username, rest = url_parts.split('@')
+    host, port_str = rest.split(':')
+    port = int(port_str)
+
+    ssh = paramiko.SSHClient()
+    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    ssh.connect(
+        hostname=host,
+        port=port,
+        username=username,
+        pkey=key
+    )
+    cleanup.add(lambda: ssh.close())
+
+    print("Creating SFTP client")
+    sftp = ssh.open_sftp()
+    cleanup.add(lambda: sftp.close())
+
+    print("Writing test file")
+    test_content = "Hello from Gitpod Python SDK!"
+    with sftp.file('test.txt', 'w') as f:
+        f.write(test_content)
+
+    with sftp.file('test.txt', 'r') as f:
+        content = f.read()
+        print(f"File content: {content.decode()}")
+
+if __name__ == "__main__":
+    disposables = util.Disposables()
+    with disposables:
+        asyncio.run(main(disposables))

examples/run_command.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python
+
+import sys
+import asyncio
+
+import gitpod.lib as util
+from gitpod import AsyncGitpod
+from gitpod.types.environment_spec_param import EnvironmentSpecParam
+from gitpod.types.environment_initializer_param import Spec
+
+
+# Examples:
+# - ./examples/run_command.py 'echo "Hello World!"'
+# - ./examples/run_command.py 'echo "Hello World!"' https://github.com/gitpod-io/empty
+async def main(cleanup: util.Disposables) -> None:
+    client = AsyncGitpod()
+
+    if len(sys.argv) < 2:
+        print("Usage: ./examples/run_command.py '<COMMAND>' [CONTEXT_URL]")
+        sys.exit(1)
+
+    command = sys.argv[1]
+    context_url = sys.argv[2] if len(sys.argv) > 2 else None
+
+    env_class = await util.find_most_used_environment_class(client)
+    if not env_class:
+        print("Error: No environment class found. Please create one first.")
+        sys.exit(1)
+    print(f"Found environment class: {env_class.display_name} ({env_class.description})")
+    env_class_id = env_class.id
+    assert env_class_id is not None
+
+    spec: EnvironmentSpecParam = {
+        "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
+        "machine": {"class": env_class_id},
+    }
+    if context_url:
+        spec["content"] = {
+            "initializer": {"specs": [Spec(
+                context_url={
+                    "url": context_url
+                }
+            )]}
+        }
+
+    print("Creating environment")
+    environment = (await client.environments.create(spec=spec)).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
+    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
+
+    print("Waiting for environment to be ready")
+    await util.wait_for_environment_ready(client, environment_id)
+
+    print("Running command")
+    lines = await util.run_command(client, environment_id, command)
+    async for line in lines:
+        print(line)
+
+if __name__ == "__main__":
+    disposables = util.Disposables()
+    with disposables:
+        asyncio.run(main(disposables))