feat(auth): Add SCM authentication support for repository access (#44)

Author: akosyakov

Diff for: examples/anthropic_tool_use.py

@@ -1,4 +1,5 @@
-#!/usr/bin/env python
+# export ANTHROPIC_API_KEY=...
+# python -m examples.anthropic_tool_use
 
 from __future__ import annotations
 
@@ -11,6 +12,8 @@
 from gitpod import AsyncGitpod
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 gpclient = AsyncGitpod()
 llmclient = Anthropic()
 
@@ -41,8 +44,7 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
     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
+    await verify_context_url(gpclient, args["context_url"], env_class.runner_id)
 
     environment = (await gpclient.environments.create(
         spec={
@@ -54,18 +56,15 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
                     }
                 )]},
             },
-            "machine": {"class": env_class_id},
+            "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)))    
+    cleanup.adda(lambda: 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
+    print(f"\nCreated environment: {environment.id} - waiting for it to be ready...")
+    await util.wait_for_environment_running(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"])
@@ -135,6 +134,4 @@ async def main(cleanup: util.Disposables) -> None:
 
 if __name__ == "__main__":
     import asyncio
-    disposables = util.Disposables()
-    with disposables:
-        asyncio.run(main(disposables))
+    asyncio.run(util.with_disposables(main))

Diff for: examples/fs_access.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
 import sys
 import asyncio
 from io import StringIO
@@ -11,10 +9,12 @@
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 
 # Examples:
-# - ./examples/fs_access.py
-# - ./examples/fs_access.py https://github.com/gitpod-io/empty
+# - python -m examples.fs_access
+# - python -m examples.fs_access https://github.com/gitpod-io/empty
 async def main(cleanup: util.Disposables) -> None:
     client = AsyncGitpod()
 
@@ -25,8 +25,6 @@ async def main(cleanup: util.Disposables) -> None:
         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)
@@ -39,13 +37,14 @@ async def main(cleanup: util.Disposables) -> None:
     key_id = "fs-access-example"
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class_id},
+        "machine": {"class": env_class.id},
         "ssh_public_keys": [{
             "id": key_id,
             "value": public_key
         }]
     }
     if context_url:
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
                 context_url={
@@ -56,13 +55,10 @@ async def main(cleanup: util.Disposables) -> None:
 
     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)))
+    cleanup.adda(lambda: client.environments.delete(environment_id=environment.id))
 
-    env = util.EnvironmentState(client, environment_id)
-    cleanup.add(lambda: asyncio.run(env.close()))
+    env = util.EnvironmentState(client, environment.id)
+    cleanup.adda(lambda: env.close())
 
     print("Waiting for environment to be running")
     await env.wait_until_running()
@@ -104,6 +100,4 @@ async def main(cleanup: util.Disposables) -> None:
         print(f"File content: {content.decode()}")
 
 if __name__ == "__main__":
-    disposables = util.Disposables()
-    with disposables:
-        asyncio.run(main(disposables))
+    asyncio.run(util.with_disposables(main))

Diff for: examples/run_command.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
 import sys
 import asyncio
 
@@ -8,10 +6,12 @@
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 
 # Examples:
-# - ./examples/run_command.py 'echo "Hello World!"'
-# - ./examples/run_command.py 'echo "Hello World!"' https://github.com/gitpod-io/empty
+# - python -m examples.run_command 'echo "Hello World!"'
+# - python -m examples.run_command 'echo "Hello World!"' https://github.com/gitpod-io/empty
 async def main(cleanup: util.Disposables) -> None:
     client = AsyncGitpod()
 
@@ -27,14 +27,13 @@ async def main(cleanup: util.Disposables) -> None:
         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},
+        "machine": {"class": env_class.id},
     }
     if context_url:
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
                 context_url={
@@ -45,20 +44,15 @@ async def main(cleanup: util.Disposables) -> None:
 
     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)))
+    cleanup.adda(lambda: client.environments.delete(environment_id=environment.id))
 
     print("Waiting for environment to be ready")
-    await util.wait_for_environment_ready(client, environment_id)
+    await util.wait_for_environment_running(client, environment.id)
 
     print("Running command")
-    lines = await util.run_command(client, environment_id, 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))
+    asyncio.run(util.with_disposables(main))

Diff for: examples/run_service.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
 import sys
 import asyncio
 
@@ -8,56 +6,53 @@
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 
 # Examples:
-# - ./examples/run_service.py
-# - ./examples/run_service.py https://github.com/gitpod-io/empty
+# - python -m examples.run_service
+# - python -m examples.run_service 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
 
     port = 8888
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class_id},
+        "machine": {"class": env_class.id},
         "ports": [{
             "name": "Lama Service",
             "port": port,
             "admission": "ADMISSION_LEVEL_EVERYONE"
         }]
     }
     if context_url:
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
-             context_url={
-                 "url": context_url
-             }
-        )]}
-    }
+                 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)))
+    cleanup.adda(lambda: client.environments.delete(environment_id=environment.id))
 
     print("Waiting for environment to be ready")
-    env = util.EnvironmentState(client, environment_id)
-    cleanup.add(lambda: asyncio.run(env.close()))
+    env = util.EnvironmentState(client, environment.id)
+    cleanup.adda(lambda: env.close())
     await env.wait_until_running()
 
     print("Starting Lama Service")
-    lines = await util.run_service(client, environment_id, {
+    lines = await util.run_service(client, environment.id, {
         "name":"Lama Service",
         "description":"Lama Service", 
         "reference":"lama-service"
@@ -75,6 +70,4 @@ async def main(cleanup: util.Disposables) -> None:
         print(line)
 
 if __name__ == "__main__":
-    disposables = util.Disposables()
-    with disposables:
-        asyncio.run(main(disposables))
+    asyncio.run(util.with_disposables(main))

Diff for: examples/scm_auth.py

@@ -0,0 +1,132 @@
+import sys
+from urllib.parse import urlparse
+
+import gitpod
+import gitpod.lib as util
+from gitpod import AsyncGitpod
+from gitpod.types.runner_check_authentication_for_host_response import SupportsPat
+
+
+async def handle_pat_auth(client: AsyncGitpod, user_id: str, runner_id: str, host: str, supports_pat: SupportsPat) -> None:
+    print("\nTo create a Personal Access Token:")
+    create_url = supports_pat.create_url
+    
+    if create_url: 
+        print(f"1. Visit: {create_url}")
+    else:
+        print(f"1. Go to {host} > Settings > Developer Settings")
+    
+    if supports_pat.required_scopes and len(supports_pat.required_scopes) > 0:
+        required_scopes = ", ".join(supports_pat.required_scopes)
+        print(f"2. Create a new token with the following scopes: {required_scopes}")
+    else:
+        print(f"2. Create a new token")
+    
+    if supports_pat.example:
+        print(f"3. Copy the generated token (example: {supports_pat.example})")
+    else:
+        print(f"3. Copy the generated token")
+    
+    if supports_pat.docs_url:
+        print(f"\nFor detailed instructions, visit: {supports_pat.docs_url}")
+
+    pat = input("\nEnter your Personal Access Token: ").strip()
+    if not pat:
+        return
+
+    await util.set_scm_pat(client, user_id, runner_id, host, pat)
+
+async def verify_context_url(client: AsyncGitpod, context_url: str, runner_id: str) -> None:
+    """Verify and handle authentication for a repository context URL.
+    
+    This function checks if the user has access to the specified repository and manages
+    the authentication process if needed. Git access to the repository is required for
+    environments to function properly.
+
+    As an alternative, you can authenticate once via the Gitpod dashboard:
+    1. Start a new environment 
+    2. Complete the browser-based authentication flow
+    
+    See https://www.gitpod.io/docs/flex/source-control for more details.
+    """
+    host = urlparse(context_url).hostname
+    if host is None:
+        print("Error: Invalid context URL")
+        sys.exit(1)
+
+    user = (await client.users.get_authenticated_user()).user
+
+    # Main authentication loop
+    first_attempt = True
+    while True:
+        try:
+            # Try to access the context URL
+            await client.runners.parse_context_url(context_url=context_url, runner_id=runner_id)
+            print("\n✓ Authentication verified successfully")
+            return
+
+        except gitpod.APIError as e:
+            if e.code != "failed_precondition":
+                raise e
+            
+        # Show authentication required message only on first attempt
+        if first_attempt:
+            print(f"\nAuthentication required for {host}")
+            first_attempt = False
+
+        # Get authentication options for the host
+        auth_resp = await client.runners.check_authentication_for_host(
+            host=host,
+            runner_id=runner_id
+        )
+
+        # Handle re-authentication case
+        if auth_resp.authenticated and not first_attempt:
+            print("\nIt looks like you are already authenticated.")
+            if input("Would you like to re-authenticate? (y/n): ").lower().strip() != 'y':
+                print("\nAuthentication cancelled")
+                sys.exit(1)
+            else:
+                print("\nRetrying authentication...")
+                continue
+                
+        auth_methods: list[tuple[str, str]] = []
+        if auth_resp.supports_oauth2:
+            auth_methods.append(("OAuth", "Recommended"))
+        if auth_resp.supports_pat:
+            auth_methods.append(("Personal Access Token (PAT)", ""))
+
+        if not auth_methods:
+            print(f"\nError: No authentication method available for {host}")
+            sys.exit(1)
+
+        # Present authentication options
+        if len(auth_methods) > 1:
+            print("\nAvailable authentication methods:")
+            for i, (method, note) in enumerate(auth_methods, 1):
+                note_text = f" ({note})" if note else ""
+                print(f"{i}. {method}{note_text}")
+            
+            choice = input(f"\nChoose authentication method (1-{len(auth_methods)}): ").strip()
+            try:
+                method_index = int(choice) - 1
+                if not 0 <= method_index < len(auth_methods):
+                    raise ValueError()
+            except ValueError:
+                method_index = 0  # Default to OAuth if invalid input
+        else:
+            method_index = 0
+
+        # Handle chosen authentication method
+        chosen_method = auth_methods[method_index][0]
+        if chosen_method == "Personal Access Token (PAT)":
+            assert auth_resp.supports_pat
+            await handle_pat_auth(client, user.id, runner_id, host, auth_resp.supports_pat)
+        else:
+            assert auth_resp.supports_oauth2
+            print(f"\nPlease visit the following URL to authenticate:")
+            print(f"{auth_resp.supports_oauth2.auth_url}")
+            if auth_resp.supports_oauth2.docs_url:
+                print(f"\nFor detailed instructions, visit: {auth_resp.supports_oauth2.docs_url}")
+            print("\nWaiting for authentication to complete...")
+            input("Press Enter after completing authentication in your browser...") 

Diff for: requirements-dev.lock

@@ -73,7 +73,6 @@ packaging==23.2
     # via nox
     # via pytest
 paramiko==3.5.1
-    # via gitpod-sdk
 platformdirs==3.11.0
     # via virtualenv
 pluggy==1.5.0

Diff for: requirements.lock

@@ -14,16 +14,9 @@ annotated-types==0.6.0
 anyio==4.4.0
     # via gitpod-sdk
     # via httpx
-bcrypt==4.2.1
-    # via paramiko
 certifi==2023.7.22
     # via httpcore
     # via httpx
-cffi==1.17.1
-    # via cryptography
-    # via pynacl
-cryptography==44.0.1
-    # via paramiko
 distro==1.8.0
     # via gitpod-sdk
 exceptiongroup==1.2.2
@@ -37,16 +30,10 @@ httpx==0.28.1
 idna==3.4
     # via anyio
     # via httpx
-paramiko==3.5.1
-    # via gitpod-sdk
-pycparser==2.22
-    # via cffi
 pydantic==2.10.3
     # via gitpod-sdk
 pydantic-core==2.27.1
     # via pydantic
-pynacl==1.5.0
-    # via paramiko
 sniffio==1.3.0
     # via anyio
     # via gitpod-sdk

Diff for: src/gitpod/lib/__init__.py

@@ -1,12 +1,21 @@
+from .runner import set_scm_pat
 from .automation import run_command, run_service
-from .disposables import Disposables
-from .environment import EnvironmentState, wait_for_environment_ready, find_most_used_environment_class
+from .disposables import Disposables, with_disposables
+from .environment import (
+    EnvironmentState,
+    find_environment_class_by_id,
+    wait_for_environment_running,
+    find_most_used_environment_class,
+)
 
 __all__ = [
     'find_most_used_environment_class',
     'run_command',
     'run_service',
     'EnvironmentState',
     'Disposables',
-    'wait_for_environment_ready',
+    'wait_for_environment_running',
+    'find_environment_class_by_id',
+    'set_scm_pat',
+    'with_disposables',
 ] 

Diff for: src/gitpod/lib/automation.py

@@ -25,21 +25,17 @@ async def run_service(
         }
     )).services
 
-    service_id: Optional[str] = None
     if not services:
         service = (await client.environments.automations.services.create(   
             environment_id=environment_id,
             spec=spec,
             metadata=metadata
         )).service
-        assert service is not None
-        service_id = service.id
     else:
-        service_id = services[0].id
-    assert service_id is not None
+        service = services[0]
 
-    await client.environments.automations.services.start(id=service_id)
-    log_url = await wait_for_service_log_url(client, environment_id, service_id)
+    await client.environments.automations.services.start(id=service.id)
+    log_url = await wait_for_service_log_url(client, environment_id, service.id)
     return stream_logs(client, environment_id, log_url)
 
 async def run_command(client: AsyncGitpod, environment_id: str, command: str) -> AsyncIterator[str]:
@@ -50,7 +46,6 @@ async def run_command(client: AsyncGitpod, environment_id: str, command: str) ->
         }
     )).tasks
 
-    task_id: Optional[str] = None
     if not tasks:
         task = (await client.environments.automations.tasks.create(
             spec={
@@ -63,23 +58,17 @@ async def run_command(client: AsyncGitpod, environment_id: str, command: str) ->
                 "reference": TASK_REFERENCE,
             },
         )).task
-        assert task is not None
-        task_id = task.id
     else:
-        task_id = tasks[0].id
-        assert task_id is not None
+        task = tasks[0]
         await client.environments.automations.tasks.update(
-            id=task_id,
+            id=task.id,
             spec={
                 "command": command,
             },
         )
-    assert task_id is not None
-    task_execution = (await client.environments.automations.tasks.start(id=task_id)).task_execution
-    assert task_execution is not None
-    task_execution_id = task_execution.id
-    assert task_execution_id is not None
-    log_url = await wait_for_task_log_url(client, environment_id, task_execution_id)
+
+    task_execution = (await client.environments.automations.tasks.start(id=task.id)).task_execution
+    log_url = await wait_for_task_log_url(client, environment_id, task_execution.id)
     return stream_logs(client, environment_id, log_url)
 
 async def wait_for_task_log_url(client: AsyncGitpod, environment_id: str, task_execution_id: str) -> str:

Diff for: src/gitpod/lib/disposables.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, List, Callable
+from typing import Any, List, Callable, Awaitable
 
 log = logging.getLogger(__name__)
 
@@ -21,30 +21,41 @@ class Disposables:
     """
 
     def __init__(self) -> None:
-        self._actions: List[Callable[[], Any]] = []
+        self._actions: List[Callable[[], Awaitable[Any]]] = []
+
+    def adda(self, action: Callable[[], Awaitable[Any]]) -> None:
+        """Add an async cleanup action to be executed when the context exits.
+        
+        Args:
+            action: An async callable that performs cleanup when called
+        """
+        self._actions.append(action)
 
     def add(self, action: Callable[[], Any]) -> None:
         """Add a cleanup action to be executed when the context exits.
         
         Args:
             action: A callable that performs cleanup when called
         """
-        self._actions.append(action)
-    
-    def __enter__(self) -> 'Disposables':
-        return self
+        async def wrapper() -> Any:
+            return action()
+        self._actions.append(wrapper)
 
-    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
-        self.cleanup()
-    
-    def cleanup(self) -> None:
+    async def cleanup(self) -> None:
         """Execute all cleanup actions in reverse order.
         
         If any cleanup action raises an exception, it will be logged but won't prevent
         other cleanup actions from executing.
         """
         for action in reversed(self._actions):
             try:
-                action()
+                await action()
             except BaseException:
-                log.exception("cleanup action failed")
+                log.exception("cleanup action failed")
+
+async def with_disposables(fn: Callable[['Disposables'], Awaitable[Any]]) -> Any:
+    disposables = Disposables()
+    try:
+        return await fn(disposables)
+    finally:
+        await disposables.cleanup()

Diff for: src/gitpod/lib/environment.py

@@ -39,7 +39,6 @@ async def _update_environment(self) -> None:
         try:
             resp = await self.client.environments.retrieve(environment_id=self.environment_id)
             env = resp.environment
-            assert env is not None
             self._environment = env
             self._ready.set()
             for listener in list(self._listeners):
@@ -135,7 +134,7 @@ def listener(env: Environment) -> None:
             await event.wait()
             if result is None:
                 raise RuntimeError("wait_until completed but result is None")
-            return result 
+            return result # type: ignore[unreachable] 
         finally:
             self._listeners.remove(listener)
 
@@ -144,11 +143,11 @@ def is_running(self, env: Environment) -> bool:
         if not env.status:
             return False
 
-        if env.status.phase in ["ENVIRONMENT_PHASE_STOPPING", "ENVIRONMENT_PHASE_STOPPED", 
+        if env.status.failure_message:
+            raise RuntimeError(f"Environment {env.id} failed: {'; '.join(env.status.failure_message)}")
+        elif env.status.phase in ["ENVIRONMENT_PHASE_STOPPING", "ENVIRONMENT_PHASE_STOPPED", 
                               "ENVIRONMENT_PHASE_DELETING", "ENVIRONMENT_PHASE_DELETED"]:
             raise RuntimeError(f"Environment {env.id} is in unexpected phase: {env.status.phase}")
-        elif env.status.failure_message:
-            raise RuntimeError(f"Environment {env.id} failed: {'; '.join(env.status.failure_message)}")
 
         return env.status.phase == "ENVIRONMENT_PHASE_RUNNING"
 
@@ -212,7 +211,7 @@ def check_key(env: Environment) -> Optional[bool]:
             return True if self.check_ssh_key_applied(env, key_id, key_value) else None
         await self.wait_until(check_key)
 
-async def wait_for_environment_ready(client: AsyncGitpod, environment_id: str) -> None:
+async def wait_for_environment_running(client: AsyncGitpod, environment_id: str) -> None:
     env = EnvironmentState(client, environment_id)
     try: 
         await env.wait_until_running()
@@ -240,6 +239,9 @@ async def find_most_used_environment_class(client: AsyncGitpod) -> Optional[Envi
     if not environment_class_id:
         return None
 
+    return await find_environment_class_by_id(client, environment_class_id)
+
+async def find_environment_class_by_id(client: AsyncGitpod, environment_class_id: str) -> Optional[EnvironmentClass]:
     classes_resp = await client.environments.classes.list(filter={"can_create_environments": True})
     while classes_resp:
         for cls in classes_resp.environment_classes:

Diff for: src/gitpod/lib/runner.py

@@ -0,0 +1,36 @@
+from gitpod._client import AsyncGitpod
+
+
+async def set_scm_pat(client: AsyncGitpod, user_id: str, runner_id: str, host: str, pat: str) -> None:
+    """Set a Personal Access Token (PAT) for source control authentication.
+
+    This will delete any existing tokens for the given host and create a new one.
+
+    Args:
+        client: The AsyncGitpod client instance
+        user_id: ID of the user to set the token for
+        runner_id: ID of the runner to associate the token with
+        host: Source control host (e.g. github.com, gitlab.com)
+        pat: The Personal Access Token string
+    """
+    tokens_response = await client.runners.configurations.host_authentication_tokens.list(
+        filter={
+            "user_id": user_id,
+            "runner_id": runner_id,
+        }
+    )
+
+    if tokens_response and tokens_response.tokens:
+        for token in tokens_response.tokens:
+            if token.host == host:
+                await client.runners.configurations.host_authentication_tokens.delete(
+                    id=token.id
+                )
+
+    await client.runners.configurations.host_authentication_tokens.create(
+        token=pat,
+        host=host,
+        runner_id=runner_id,
+        user_id=user_id,
+        source="HOST_AUTHENTICATION_TOKEN_SOURCE_PAT"
+    )