refactor, handle zombie tasks

Bump pyworker version

.gitignore

@@ -2,5 +2,4 @@
 .envrc
 __pycache__
 bin/
-lib64
-.venv
+lib64

README.md

@@ -39,12 +39,11 @@ reporting these metrics to the autoscaler.
 
 If you are using a Vast.ai template that includes PyWorker integration (marked as autoscaler compatible), it should work out of the box. The template will typically start the appropriate PyWorker server automatically. Here's a few:
 
-*   **vLLM:** [Vast.ai Template](https://cloud.vast.ai?ref_id=62897&template_id=63ae93902bf3978bea033782592b784d)
-*   **TGI (Text Generation Inference):** [Vast.ai Template](https://cloud.vast.ai?ref_id=62897&template_id=6fa6bd5bdf5f0df63db80e40b086037d)
-*   **ComfyUI:** [Vast.ai Template](https://cloud.vast.ai?ref_id=62897&template_id=e6748878ba688e765e3e9fca29541938)
+*   **TGI (Text Generation Inference):** [Vast.ai Template](https://cloud.vast.ai?ref_id=140778&template_id=72d8dcb41ea3a58e06c741e2c725bc00)
+*   **ComfyUI:** [Vast.ai Template](https://cloud.vast.ai?ref_id=140778&template_id=ad72c8bf7cf695c3c9ddf0eaf6da0447)
 
 Currently available workers:
-*   `openai`: A simple example worker for a basic vLLM server.
+*   `hello_world`: A simple example worker for a basic LLM server.
 *   `comfyui`: A worker for the ComfyUI image generation backend.
 *   `tgi`: A worker for the Text Generation Inference backend.
 

lib/backend.py

@@ -9,6 +9,7 @@ from asyncio import wait, sleep, gather, Semaphore, FIRST_COMPLETED, create_task
 from typing import Tuple, Awaitable, NoReturn, List, Union, Callable, Optional
 from functools import cached_property
 from distutils.util import strtobool
+from collections import deque
 
 from anyio import open_file
 from aiohttp import web, ClientResponse, ClientSession, ClientConnectorError, ClientTimeout, TCPConnector
@@ -63,6 +64,7 @@ class Backend:
     version = VERSION
     msg_history = []
     sem: Semaphore = dataclasses.field(default_factory=Semaphore)
+    queue: deque = dataclasses.field(default_factory=deque, repr=False)
     unsecured: bool = dataclasses.field(
         default_factory=lambda: bool(strtobool(os.environ.get("UNSECURED", "false"))),
     )
@@ -141,11 +143,26 @@ class Backend:
         workload = payload.count_workload()
         request_metrics: RequestMetrics = RequestMetrics(request_idx=auth_data.request_idx, reqnum=auth_data.reqnum, workload=workload, status="Created")
 
-        async def cancel_api_call_if_disconnected() -> web.Response:
+
+        def advance_queue_after_completion(event: asyncio.Event):
+            """Pop current head and wake next waiter, if any."""
+            # If this event is current head, wake next waiter
+            if self.queue and self.queue[0] is event:
+                self.queue.popleft()
+                if self.queue:
+                    self.queue[0].set()
+            else:
+                # Else, remove it from the queue
+                try:
+                    self.queue.remove(event)
+                except ValueError:
+                    pass
+
+        async def cancel_api_call_if_disconnected() -> None:
             await request.wait_for_disconnection()
-            log.debug(f"request with reqnum: {request_metrics.reqnum} was canceled")
+            log.debug(f"Request with reqnum: {request_metrics.reqnum} was canceled")
             self.metrics._request_canceled(request_metrics)
-            raise asyncio.CancelledError
+            return
 
         async def make_request() -> Union[web.Response, web.StreamResponse]:
             try:
@@ -162,7 +179,9 @@ class Backend:
                 res = await handler.generate_client_response(request, response)
                 self.metrics._request_success(request_metrics)
                 return res
-            except requests.exceptions.RequestException as e:
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
                 log.debug(f"[backend] Request error: {e}")
                 self.metrics._request_errored(request_metrics)
                 return web.Response(status=500)
@@ -177,46 +196,87 @@ class Backend:
             self.metrics._request_reject(request_metrics)
             return web.Response(status=429)
 
-        acquired = False
-        try:
-            self.metrics._request_start(request_metrics)
-            if self.allow_parallel_requests is False:
-                log.debug(f"Waiting to aquire Sem for reqnum:{request_metrics.reqnum}")
-                await self.sem.acquire()
-                acquired = True
-                log.debug(
-                    f"Sem acquired for reqnum:{request_metrics.reqnum}, starting request..."
-                )
-            else:
-                log.debug(f"Starting request for reqnum:{request_metrics.reqnum}")
-            done, pending = await wait(
-                [
-                    create_task(make_request()),
-                    create_task(cancel_api_call_if_disconnected()),
-                ],
-                return_when=FIRST_COMPLETED,
-            )
-            for t in pending:
-                t.cancel()
-            await asyncio.gather(*pending, return_exceptions=True)
+        disconnect_task = create_task(cancel_api_call_if_disconnected())
+        next_request_task = None
+        work_task = None
+        event = asyncio.Event() # Used in finally block, so initialize here
 
-            done_task = done.pop()
-            try:
-                return done_task.result()
-            except Exception as e:
-                log.debug(f"Request task raised exception: {e}")
-                return web.Response(status=500)
+        self.metrics._request_start(request_metrics)
+
+        try:
+            if self.allow_parallel_requests:
+                log.debug(f"Starting request for reqnum:{request_metrics.reqnum}")
+                work_task = create_task(make_request())
+                done, pending = await wait([work_task, disconnect_task], return_when=FIRST_COMPLETED)
+
+                for t in pending:
+                    t.cancel()
+                await asyncio.gather(*pending, return_exceptions=True)
+
+                if disconnect_task in done:
+                    return web.Response(status=499)
+
+                # otherwise work_task completed
+                return await work_task
+            
+            # FIFO-queue branch
+            else:
+                # Insert a Event into the queue for this request
+                # Event.set() == our request is up next
+                self.queue.append(event)
+                if self.queue and self.queue[0] is event:
+                    event.set()
+
+                # Race between our request being next and request being cancelled
+                next_request_task = create_task(event.wait())
+                first_done, first_pending = await wait(
+                    [next_request_task, disconnect_task], return_when=FIRST_COMPLETED
+                )
+
+                # If the disconnect task wins the race
+                if disconnect_task in first_done:
+                    # Clean up the next_request_task, then exit
+                    for t in first_pending:
+                        t.cancel()
+                    await asyncio.gather(*first_pending, return_exceptions=True)
+                    return web.Response(status=499)
+
+                # We are the next-up request in the queue
+                log.debug(f"Starting work on request {request_metrics.reqnum}...")
+
+                # Race the backend API call with the disconnect task
+                work_task = create_task(make_request())
+
+                done, pending = await wait([work_task, disconnect_task], return_when=FIRST_COMPLETED)
+                for t in pending:
+                    t.cancel()
+                await asyncio.gather(*pending, return_exceptions=True)
+
+                if disconnect_task in done:
+                    return web.Response(status=499)
+
+                # otherwise work_task completed
+                return await work_task
+            
         except asyncio.CancelledError:
-            # Client is gone. Do not write a response; just unwind.
-            return web.Response(status=499) 
+            return web.Response(status=499)
+        
         except Exception as e:
             log.debug(f"Exception in main handler loop {e}")
             return web.Response(status=500)
+        
         finally:
-            # Always release the semaphore if it was acquired
-            if acquired:
-                self.sem.release()
+            if not self.allow_parallel_requests:
+                advance_queue_after_completion(event)
+
             self.metrics._request_end(request_metrics)
+            cleanup_tasks = [t for t in (next_request_task, work_task, disconnect_task) if t]
+            for t in cleanup_tasks:
+                if not t.done():
+                    t.cancel()
+            if cleanup_tasks:
+                await asyncio.gather(*cleanup_tasks, return_exceptions=True)
+
 
     @cached_property  
     def healthcheck_session(self):
@@ -417,7 +477,7 @@ class Backend:
 
         async def tail_log():
             log.debug(f"tailing file: {self.model_log_file}")
-            async with await open_file(self.model_log_file, encoding='utf-8', errors='ignore') as f:
+            async with await open_file(self.model_log_file) as f:
                 while True:
                     line = await f.readline()
                     if line:

lib/server.py

@@ -3,58 +3,38 @@ import logging
 from typing import List
 import ssl
 from asyncio import run, gather
-import asyncio
+
 
 from lib.backend import Backend
-from lib.metrics import Metrics
 from aiohttp import web
 
 log = logging.getLogger(__file__)
 
 
 def start_server(backend: Backend, routes: List[web.RouteDef], **kwargs):
-    try:
-        log.debug("getting certificate...")
-        use_ssl = os.environ.get("USE_SSL", "false") == "true"
-        if use_ssl is True:
-            ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
-            ssl_context.load_cert_chain(
-                certfile="/etc/instance.crt",
-                keyfile="/etc/instance.key",
-            )
-        else:
-            ssl_context = None
+    log.debug("getting certificate...")
+    use_ssl = os.environ.get("USE_SSL", "false") == "true"
+    if use_ssl is True:
+        ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        ssl_context.load_cert_chain(
+            certfile="/etc/instance.crt",
+            keyfile="/etc/instance.key",
+        )
+    else:
+        ssl_context = None
 
-        async def main():
-            log.debug("starting server...")
-            app = web.Application()
-            app.add_routes(routes)
-            runner = web.AppRunner(app)
-            await runner.setup()
-            site = web.TCPSite(
-                runner,
-                ssl_context=ssl_context,
-                port=int(os.environ["WORKER_PORT"]),
-                **kwargs
-            )
-            await gather(site.start(), backend._start_tracking())
+    async def main():
+        log.debug("starting server...")
+        app = web.Application()
+        app.add_routes(routes)
+        runner = web.AppRunner(app)
+        await runner.setup()
+        site = web.TCPSite(
+            runner,
+            ssl_context=ssl_context,
+            port=int(os.environ["WORKER_PORT"]),
+            **kwargs
+        )
+        await gather(site.start(), backend._start_tracking())
 
-        run(main())
-
-    except Exception as e:
-        err_msg = f"PyWorker failed to launch: {e}"
-        log.error(err_msg)
-
-        async def beacon():
-            metrics = Metrics()
-            metrics._set_version(getattr(backend, "version", "0"))
-            metrics._set_mtoken(getattr(backend, "mtoken", ""))
-            try:
-                while True:
-                    metrics._model_errored(err_msg)
-                    await metrics._Metrics__send_metrics_and_reset()
-                    await asyncio.sleep(10)
-            finally:
-                await metrics.aclose()
-
-        run(beacon())
+    run(main())

requirements.txt

@@ -8,4 +8,3 @@ Requests~=2.32
 transformers~=4.52
 utils==1.0.*
 hf_transfer>=0.1.9
-vastai-sdk>=0.2.0

start_server.sh

@@ -41,14 +41,6 @@ echo_var DEBUG_LOG
 echo_var PYWORKER_LOG
 echo_var MODEL_LOG
 
-# if instance is rebooted, we want to clear out the log file so pyworker doesn't read lines
-# from the run prior to reboot. past logs are saved in $MODEL_LOG.old for debugging only
-if [ -e "$MODEL_LOG" ]; then
-    echo "Rotating model log at $MODEL_LOG to $MODEL_LOG.old"
-    cat "$MODEL_LOG" >> "$MODEL_LOG.old" 
-    : > "$MODEL_LOG"
-fi
-
 # Populate /etc/environment with quoted values
 if ! grep -q "VAST" /etc/environment; then
     env -0 | grep -zEv "^(HOME=|SHLVL=)|CONDA" | while IFS= read -r -d '' line; do
@@ -132,43 +124,9 @@ cd "$SERVER_DIR"
 
 echo "launching PyWorker server"
 
-set +e
-python3 -m "workers.$BACKEND.server" |& tee -a "$PYWORKER_LOG"
-PY_STATUS=${PIPESTATUS[0]}
-set -e
+# if instance is rebooted, we want to clear out the log file so pyworker doesn't read lines
+# from the run prior to reboot. past logs are saved in $MODEL_LOG.old for debugging only
+[ -e "$MODEL_LOG" ] && cat "$MODEL_LOG" >> "$MODEL_LOG.old" && : > "$MODEL_LOG"
 
-if [ "${PY_STATUS}" -ne 0 ]; then
-  echo "PyWorker exited with status ${PY_STATUS}; notifying autoscaler..."
-  ERROR_MSG="PyWorker exited: code ${PY_STATUS}"
-  MTOKEN="${MASTER_TOKEN:-}"
-  VERSION="${PYWORKER_VERSION:-0}"
-
-  IFS=',' read -r -a REPORT_ADDRS <<< "${REPORT_ADDR}"
-  for addr in "${REPORT_ADDRS[@]}"; do
-    curl -sS -X POST -H 'Content-Type: application/json' \
-      -d "$(cat <<JSON
-{
-  "id": ${CONTAINER_ID:-0},
-  "mtoken": "${MTOKEN}",
-  "version": "${VERSION}",
-  "loadtime": 0,
-  "new_load": 0,
-  "cur_load": 0,
-  "rej_load": 0,
-  "max_perf": 0,
-  "cur_perf": 0,
-  "error_msg": "${ERROR_MSG}",
-  "num_requests_working": 0,
-  "num_requests_recieved": 0,
-  "additional_disk_usage": 0,
-  "working_request_idxs": [],
-  "cur_capacity": 0,
-  "max_capacity": 0,
-  "url": "${URL}"
-}
-JSON
-)" "${addr%/}/worker_status/" || true
-  done
-fi
-
-echo "launching PyWorker server done"
+(python3 -m "workers.$BACKEND.server" |& tee -a "$PYWORKER_LOG") &
+echo "launching PyWorker server done"

workers/comfyui-json/client.py

@@ -1,35 +1,156 @@
-from .data_types import count_workload
+import logging
 import uuid
 import random
-import asyncio
-import random
+from urllib.parse import urljoin
+import json
 
-from vastai import Serverless
+import requests
 
-async def main():
-    async with Serverless() as client:
-        endpoint = await client.get_endpoint(name="my-comfy-endpoint") # Change this to your endpoint name
+from lib.test_utils import print_truncate_res
+from utils.endpoint_util import Endpoint
+from utils.ssl import get_cert_file_path
+from .data_types import count_workload
 
-        payload = {
-            "input": {
-                "request_id": str(uuid.uuid4()),
-                "modifier": "Text2Image",
-                "modifications": {
-                    "prompt": "a beautiful landscape with mountains and lakes",
-                    "width": 1024,
-                    "height": 1024,
-                    "steps": 20,
-                    "seed": random.randint(0, 2**32 - 1)
-                },
-                "workflow_json": {}  # Empty since using modifier approach
-            }
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s[%(levelname)-5s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+log = logging.getLogger(__file__)
+
+
+def call_text2image_workflow(
+    endpoint_group_name: str, api_key: str, server_url: str
+) -> None:
+    """Simple Text2Image using the new modifier-based approach"""
+    
+    def make_request(url: str, payload: dict, timeout: int = None, verify=True, context: str = "request"):
+        """Helper function for making requests with consistent error handling"""
+        try:
+            response = requests.post(
+                url,
+                json=payload,
+                timeout=timeout,
+                verify=verify
+            )
+            
+            response.raise_for_status()
+            return response.json()
+            
+        except requests.exceptions.HTTPError as http_err:
+            log.error(f"HTTP error occurred during {context}: {http_err}")
+            log.error(f"Status Code: {response.status_code}")
+            log.error("Response content:", response.text)
+            return None
+        except requests.exceptions.Timeout:
+            log.error(f"Timeout occurred during {context}: {url}")
+            return None
+        except requests.exceptions.ConnectionError:
+            log.error(f"Connection error occurred during {context}: {url}")
+            return None
+        except json.JSONDecodeError as json_err:
+            log.error(f"Failed to decode JSON response during {context}: {json_err}")
+            if 'response' in locals():
+                print("Response content:", response.text)
+            return None
+        except Exception as err:
+            log.error(f"An unexpected error occurred during {context}: {err}")
+            if 'response' in locals():
+                log.error("Response content (if available):", response.text)
+            return None
+    
+    WORKER_ENDPOINT = "/generate/sync"
+
+    # This worker has concurrency = 1.  All workloads have cost value 1.0
+    COST = count_workload()
+    
+    # Route to get worker URL
+    route_payload = {
+        "endpoint": endpoint_group_name,
+        "api_key": api_key,
+        "cost": COST,
+    }
+    
+    # First request - get routing information
+    route_response = make_request(
+        url=urljoin(server_url, "/route/"),
+        payload=route_payload,
+        timeout=4,
+        context="route request"
+    )
+    
+    if route_response is None:
+        return None
+    
+    if "url" not in route_response or not route_response["url"]:
+        log.error("Error: No worker in 'Ready' state. Please wait while the serverless engine removes errored workers or finishes loading new workers.")
+        return None
+    
+    if "status" in route_response:
+        print(f"Autoscaler status: {route_response['status']}")
+        return None
+    
+    # Extract data from route response
+    url = route_response["url"]
+    auth_data = dict(
+        signature=route_response["signature"],
+        cost=route_response["cost"],
+        endpoint=route_response["endpoint"],
+        reqnum=route_response["reqnum"],
+        url=route_response["url"],
+        request_idx=route_response["request_idx"],
+    )
+    
+    # Build the payload for the worker request
+    worker_payload = {
+        "input": {
+            "request_id": str(uuid.uuid4()),
+            "modifier": "Text2Image",
+            "modifications": {
+                "prompt": "a beautiful landscape with mountains and lakes",
+                "width": 1024,
+                "height": 1024,
+                "steps": 20,
+                "seed": random.randint(0, 2**32 - 1)
+            },
+            "workflow_json": {}  # Empty since using modifier approach
         }
-        
-        response = await endpoint.request("/generate/sync", payload, cost=count_workload())
+    }
+    
+    req_data = dict(payload=worker_payload, auth_data=auth_data)
+    worker_url = urljoin(url, WORKER_ENDPOINT)
+    print(f"url: {worker_url}")
+    
+    # Second request - call the worker endpoint
+    worker_response = make_request(
+        url=worker_url,
+        payload=req_data,
+        verify=get_cert_file_path(),
+        context="worker request"
+    )
+    
+    return worker_response
 
-        # Get the file from the path on the local machine using SCP or SFTP
-        # or configure S3 to upload to cloud storage.
-        print(response["response"]["output"][0]["local_path"])
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    from lib.test_utils import test_args
+
+    args = test_args.parse_args()
+    endpoint_api_key = Endpoint.get_endpoint_api_key(
+        endpoint_name=args.endpoint_group_name,
+        account_api_key=args.api_key,
+        instance=args.instance,
+    )
+
+    if endpoint_api_key:
+        result = call_text2image_workflow(
+            api_key=endpoint_api_key,
+            endpoint_group_name=args.endpoint_group_name,
+            server_url=args.server_url,
+        )
+        if result is None:
+            log.error("Text2Image workflow failed")
+        else:   
+            print(result)
+    else:
+        log.error(f"Failed to get API key for endpoint {args.endpoint_group_name}")

workers/comfyui/client.py

@@ -7,13 +7,20 @@ from lib.test_utils import print_truncate_res
 from utils.endpoint_util import Endpoint
 from utils.ssl import get_cert_file_path
 
-from vastai import Serverless
+"""
+NOTE: this client example uses a custom comfy workflow compatible with SD3 only
+"""
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s[%(levelname)-5s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+log = logging.getLogger(__file__)
 
 
-ENDPOINT_NAME = "my-comfyui-endpoint"
-COST = 100 # Use a constant cost for image generation
-
-def call_default_workflow(client: Serverless) -> None:
+def call_default_workflow(
+    endpoint_group_name: str, api_key: str, server_url: str
+) -> None:
     WORKER_ENDPOINT = "/prompt"
     COST = 100
     route_payload = {

workers/openai/README.md

@@ -8,13 +8,14 @@ This is the base PyWorker for OpenAI compatible inference servers.  See the [Ser
 
 This worker is compatible with any backend API that properly implements the `/v1/completions` and `/v1/chat/completions` endpoints.  We currently have three templates you can choose from but you can also create your own without having to modify the PyWorker.
 
-- [vLLM](https://cloud.vast.ai/?ref_id=62897&creator_id=62897&name=vLLM%20(Serverless)) (recommended)
+- [vLLM](https://cloud.vast.ai/?ref_id=62897&creator_id=62897&name=vLLM%20%2B%20Qwen%2FQwen3-8B%20(Serverless)) (recommended)
 - [Ollama](https://cloud.vast.ai/?ref_id=62897&creator_id=62897&name=Ollama%20%2B%20Qwen3%3A32b%20(Serverless))
+- [HuggingFace TGI](https://cloud.vast.ai/?ref_id=62897&creator_id=62897&name=TGI%20%2B%20Qwen3-8B%20(Serverless))
 
 
 All of these templates can be configured via the template interface.  You may want to change the model or startup arguments, depending on the template you selected.
 
-2. Follow the [getting started guide](https://docs.vast.ai/documentation/serverless/quickstart) for help with configuring your serverless setup.  For testing, we recommend that you use the default options presented by the web interface.
+2. Follow the [getting started guide](https://docs.vast.ai/serverless/getting-started) for help with configuring your serverless setup.  For testing, we recommend that you use the default options presented by the web interface.
 
 ## Client Setup (Demo)
 
@@ -33,30 +34,12 @@ uv pip install -r requirements.txt
 
 Several examples have been provided in the client to help you get started with your own implementation.
 
-First, set your API key as an environment variable:
+### Completions
+
+Call to `/v1/completions` with json response
 
 ```bash
-export VAST_API_KEY=<your_api_key>
-```
-
-The `--model` and `--endpoint` flags are optional. If not provided, they default to `Qwen/Qwen3-8B` and `my-vllm-endpoint` respectively.
-
-### Chat Completion (streaming)
-
-Call to `/v1/chat/completions` with streaming response
-
-```bash
-python -m workers.openai.client --chat-stream --endpoint <ENDPOINT_NAME> --model <MODEL_NAME>
-```
-
-### Interactive Chat (streaming)
-
-Interactive session with calls to `/v1/chat/completions`.
-
-Type `clear` to clear the chat history or `quit` to exit.
-
-```bash
-python -m workers.openai.client --interactive --endpoint <ENDPOINT_NAME> --model <MODEL_NAME>
+python -m workers.openai.client -k <API_KEY> -e <ENDPOINT_NAME> --completion --model <MODEL_NAME>
 ```
 
 ### Chat Completion (json)
@@ -64,7 +47,15 @@ python -m workers.openai.client --interactive --endpoint <ENDPOINT_NAME> --model
 Call to `/v1/chat/completions` with json response
 
 ```bash
-python -m workers.openai.client --chat --endpoint <ENDPOINT_NAME> --model <MODEL_NAME>
+python -m workers.openai.client -k <API_KEY> -e <ENDPOINT_NAME> --chat --model <MODEL_NAME>
+```
+
+### Chat Completion (streaming)
+
+Call to `/v1/chat/completions` with streaming response
+
+```bash
+python -m workers.openai.client -k <API_KEY> -e <ENDPOINT_NAME> --chat-stream --model <MODEL_NAME>
 ```
 
 ### Tool Use (json)
@@ -74,14 +65,16 @@ Call to `/v1/chat/completions` with tool and json response.
 This test defines a simple tool which will list the contents of the local pyworker directory.  The output is then analysed by the model.
 
 ```bash
-python -m workers.openai.client --tools --endpoint <ENDPOINT_NAME> --model <MODEL_NAME>
+python -m workers.openai.client -k <API_KEY> -e <ENDPOINT_NAME> --tools --model <MODEL_NAME>
 ```
 
-### Completions
+### Interactive Chat (streaming)
 
-Call to `/v1/completions` with json response
+Interactive session with calls to `/v1/chat/completions`.
+
+Type `clear` to clear the chat history or `quit` to exit.
 
 ```bash
-python -m workers.openai.client --completion --endpoint <ENDPOINT_NAME> --model <MODEL_NAME>
+python -m workers.openai.client -k <API_KEY> -e <ENDPOINT_NAME> --interactive --model <MODEL_NAME>
 ```
 

workers/openai/client.py

@@ -1,15 +1,14 @@
 import logging
-import json
-import os
 import sys
+import json
 import subprocess
-import argparse
-from typing import Any, Dict, List, Optional
+from urllib.parse import urljoin
+from typing import Dict, Any, Optional, Iterator, Union, List
+import requests
+from utils.endpoint_util import Endpoint
+from utils.ssl import get_cert_file_path
+from .data_types.client import CompletionConfig, ChatCompletionConfig
 
-from vastai import Serverless
-import asyncio
-
-# ---------------------- Logging ----------------------
 logging.basicConfig(
     level=logging.DEBUG,
     format="%(asctime)s[%(levelname)-5s] %(message)s",
@@ -17,20 +16,135 @@ logging.basicConfig(
 )
 log = logging.getLogger(__file__)
 
-# ---------------------- Prompts ----------------------
-COMPLETIONS_PROMPT = "Zebras are primarily grazers and can subsist on lower-quality vegetation. They are preyed on mainly by"
+COMPLETIONS_PROMPT = "the capital of USA is"
 CHAT_PROMPT = "Think step by step: Tell me about the Python programming language."
-TOOLS_PROMPT = (
-    "Can you list the files in the current working directory and tell me what you see? "
-    "What do you think this directory might be for?"
-)
+TOOLS_PROMPT = "Can you list the files in the current working directory and tell me what you see? What do you think this directory might be for?"
+
+
+class APIClient:
+    """Lightweight client focused solely on API communication"""
+
+    # Remove the generic WORKER_ENDPOINT since we're now going direct
+    DEFAULT_COST = 100
+    DEFAULT_TIMEOUT = 4
+
+    def __init__(
+        self,
+        endpoint_group_name: str,
+        api_key: str,
+        server_url: str,
+        endpoint_api_key: str,
+    ):
+        self.endpoint_group_name = endpoint_group_name
+        self.api_key = api_key
+        self.server_url = server_url
+        self.endpoint_api_key = endpoint_api_key
+
+    def _get_worker_url(self, cost: int = DEFAULT_COST) -> Dict[str, Any]:
+        """Get worker URL and auth data from routing service"""
+        if not self.endpoint_api_key:
+            raise ValueError("No valid endpoint API key available")
+
+        route_payload = {
+            "endpoint": self.endpoint_group_name,
+            "api_key": self.endpoint_api_key,
+            "cost": cost,
+        }
+
+        response = requests.post(
+            urljoin(self.server_url, "/route/"),
+            json=route_payload,
+            timeout=self.DEFAULT_TIMEOUT,
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def _create_auth_data(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """Create auth data from routing response"""
+        return {
+            "signature": message["signature"],
+            "cost": message["cost"],
+            "endpoint": message["endpoint"],
+            "reqnum": message["reqnum"],
+            "url": message["url"],
+        }
+
+    def _make_request(
+        self,
+        payload: Dict[str, Any],
+        endpoint: str,
+        method: str = "POST",
+        stream: bool = False,
+    ) -> Union[Dict[str, Any], Iterator[str]]:
+        """Make request directly to the specific worker endpoint"""
+        # Get worker URL and auth data
+        cost = payload.get("max_tokens", self.DEFAULT_COST)
+        message = self._get_worker_url(cost=cost)
+        worker_url = message["url"]
+        auth_data = self._create_auth_data(message)
+
+        req_data = {"payload": {"input": payload}, "auth_data": auth_data}
+
+        url = urljoin(worker_url, endpoint)
+        log.debug(f"Making direct request to: {url}")
+        log.debug(f"Payload: {req_data}")
+
+        # Make the request using the specified method
+        if method.upper() == "POST":
+            response = requests.post(
+                url, json=req_data, stream=stream, verify=get_cert_file_path()
+            )
+        elif method.upper() == "GET":
+            response = requests.get(
+                url, params=req_data, stream=stream, verify=get_cert_file_path()
+            )
+        else:
+            raise ValueError(f"Unsupported HTTP method: {method}")
+
+        response.raise_for_status()
+
+        if stream:
+            return self._handle_streaming_response(response)
+        else:
+            return response.json()
+
+    def _handle_streaming_response(self, response: requests.Response) -> Iterator[str]:
+        """Handle streaming response and yield tokens"""
+        try:
+            for line in response.iter_lines(decode_unicode=True):
+                if line:
+                    if line.startswith("data: "):
+                        data_str = line[6:]
+                        if data_str.strip() == "[DONE]":
+                            break
+                        try:
+                            data = json.loads(data_str)
+                            yield data  # Yield the full chunk
+                        except json.JSONDecodeError:
+                            continue
+        except Exception as e:
+            log.error(f"Error handling streaming response: {e}")
+            raise
+
+    def call_completions(
+        self, config: CompletionConfig
+    ) -> Union[Dict[str, Any], Iterator[str]]:
+        payload = config.to_dict()
+
+        return self._make_request(
+            payload=payload, endpoint="/v1/completions", stream=config.stream
+        )
+
+    def call_chat_completions(
+        self, config: ChatCompletionConfig
+    ) -> Union[Dict[str, Any], Iterator[str]]:
+        payload = config.to_dict()
+
+        return self._make_request(
+            payload=payload, endpoint="/v1/chat/completions", stream=config.stream
+        )
 
-ENDPOINT_NAME = "my-vllm-endpoint"       # change this to your vLLM endpoint name
-DEFAULT_MODEL = "Qwen/Qwen3-8B"          # must support tool calling
-MAX_TOKENS = 1024
-DEFAULT_TEMPERATURE = 0.7
 
-# ---------------------- Tooling ----------------------
 class ToolManager:
     """Handles tool definitions and execution"""
 
@@ -50,7 +164,7 @@ class ToolManager:
 
     @staticmethod
     def get_ls_tool_definition() -> List[Dict[str, Any]]:
-        """OpenAI-compatible tool schema"""
+        """Get the ls tool definition"""
         return [
             {
                 "type": "function",
@@ -64,228 +178,98 @@ class ToolManager:
 
     def execute_tool_call(self, tool_call: Dict[str, Any]) -> str:
         """Execute a tool call and return the result"""
-        function_name = (tool_call.get("function") or {}).get("name")
+        function_name = tool_call["function"]["name"]
+
         if function_name == "list_files":
             return self.list_files()
-        raise ValueError(f"Unknown tool function: {function_name}")
+        else:
+            raise ValueError(f"Unknown tool function: {function_name}")
 
 
-# ----- Helpers to handle streamed tool_calls assembly -----
-def _merge_tool_call_delta(state: Dict[int, Dict[str, Any]], tc_delta: Dict[str, Any]) -> None:
-    """
-    OpenAI-style streaming sends partial tool_calls with an index and partial fields.
-    We merge into a per-index state dict until the assistant message finishes.
-    """
-    idx = tc_delta.get("index")
-    if idx is None:
-        return
-
-    entry = state.setdefault(idx, {"id": None, "function": {"name": None, "arguments": ""}, "type": "function"})
-
-    if tc_delta.get("id"):
-        entry["id"] = tc_delta["id"]
-
-    fn_delta = tc_delta.get("function") or {}
-    if "name" in fn_delta and fn_delta["name"]:
-        entry["function"]["name"] = fn_delta["name"]
-    if "arguments" in fn_delta and fn_delta["arguments"]:
-        entry["function"]["arguments"] += fn_delta["arguments"]
-
-
-def _tool_state_to_message_tool_calls(state: Dict[int, Dict[str, Any]]) -> List[Dict[str, Any]]:
-    return [state[i] for i in sorted(state.keys())]
-
-
-# ---- OpenAI-compatible calls (non-streaming) ----
-async def call_completions(client: Serverless, *, model: str, prompt: str, endpoint_name: str, **kwargs) -> Dict[str, Any]:
-
-    endpoint = await client.get_endpoint(name=endpoint_name)
-
-    payload = {
-        "input": {
-            "model": model,
-            "prompt": prompt,
-            "max_tokens": kwargs.get("max_tokens", MAX_TOKENS),
-            "temperature": kwargs.get("temperature", DEFAULT_TEMPERATURE),
-        }
-    }
-    log.debug("POST /v1/completions %s", json.dumps(payload)[:500])
-    resp = await endpoint.request("/v1/completions", payload, cost=payload["input"]["max_tokens"])
-    return resp["response"]
-
-async def call_chat_completions(client: Serverless, *, model: str, messages: List[Dict[str, Any]], endpoint_name: str, **kwargs) -> Dict[str, Any]:
-
-    endpoint = await client.get_endpoint(name=endpoint_name)
-
-    payload = {
-        "input": {
-            "model": model,
-            "messages": messages,
-            "max_tokens": kwargs.get("max_tokens", MAX_TOKENS),
-            "temperature": kwargs.get("temperature", DEFAULT_TEMPERATURE),
-            **({"tools": kwargs["tools"]} if "tools" in kwargs else {}),
-            **({"tool_choice": kwargs["tool_choice"]} if "tool_choice" in kwargs else {}),
-        }
-    }
-    log.debug("POST /v1/chat/completions %s", json.dumps(payload)[:500])
-    resp = await endpoint.request("/v1/chat/completions", payload, cost=payload["input"]["max_tokens"])
-    return resp["response"]
-
-# ---- Streaming variants ----
-async def stream_completions(client: Serverless, *, model: str, prompt: str, endpoint_name: str, **kwargs):
-
-    endpoint = await client.get_endpoint(name=endpoint_name)
-
-    payload = {
-        "input": {
-            "model": model,
-            "prompt": prompt,
-            "max_tokens": kwargs.get("max_tokens", MAX_TOKENS),
-            "temperature": kwargs.get("temperature", DEFAULT_TEMPERATURE),
-            "stream": True,
-            **({"stop": kwargs["stop"]} if "stop" in kwargs else {}),
-        }
-    }
-    log.debug("STREAM /v1/completions %s", json.dumps(payload)[:500])
-    resp = await endpoint.request("/v1/completions", payload, cost=payload["input"]["max_tokens"], stream=True)
-    return resp["response"]  # async generator
-
-async def stream_chat_completions(client: Serverless, *, model: str, messages: List[Dict[str, Any]], endpoint_name: str, **kwargs):
-
-    endpoint = await client.get_endpoint(name=endpoint_name)
-
-    payload = {
-        "input": {
-            "model": model,
-            "messages": messages,
-            "max_tokens": kwargs.get("max_tokens", MAX_TOKENS),
-            "temperature": kwargs.get("temperature", DEFAULT_TEMPERATURE),
-            "stream": True,
-            **({"tools": kwargs["tools"]} if "tools" in kwargs else {}),
-            **({"tool_choice": kwargs["tool_choice"]} if "tool_choice" in kwargs else {}),
-        }
-    }
-    log.debug("STREAM /v1/chat/completions %s", json.dumps(payload)[:500])
-    resp = await endpoint.request("/v1/chat/completions", payload, cost=payload["input"]["max_tokens"], stream=True)
-    return resp["response"]  # async generator
-
-
-# ---------------------- Demo Runner ----------------------
 class APIDemo:
     """Demo and testing functionality for the API client"""
 
-    def __init__(self, client: Serverless, model: str, endpoint_name: str, tool_manager: Optional[ToolManager] = None):
+    def __init__(
+        self, client: APIClient, model: str, tool_manager: Optional[ToolManager] = None
+    ):
         self.client = client
         self.model = model
-        self.endpoint_name = endpoint_name
         self.tool_manager = tool_manager or ToolManager()
 
-    # ----- Streaming handler -----
-    async def handle_streaming_response(self, stream, show_reasoning: bool = True) -> str:
+    def handle_streaming_response(
+        self, response_stream, show_reasoning: bool = True
+    ) -> str:
+        """
+        Handle streaming chat response and display all output.
+        """
+
         full_response = ""
         reasoning_content = ""
-        printed_reasoning = False
-        printed_answer = False
-        finish_reason = None
+        reasoning_started = False
+        content_started = False
 
-        async for chunk in stream:
-            choice = (chunk.get("choices") or [{}])[0]
-            delta = choice.get("delta", {})
-            
-            # Track finish reason
-            if choice.get("finish_reason"):
-                finish_reason = choice.get("finish_reason")
+        for chunk in response_stream:
+            # Normalize the chunk
+            if isinstance(chunk, str):
+                chunk = chunk.strip()
+                if chunk.startswith("data: "):
+                    chunk = chunk[6:].strip()
+                if chunk in ["[DONE]", ""]:
+                    continue
+                try:
+                    parsed_chunk = json.loads(chunk)
+                except json.JSONDecodeError:
+                    continue
+            elif isinstance(chunk, dict):
+                parsed_chunk = chunk
+            else:
+                continue
 
-            # reasoning tokens
-            rc = delta.get("reasoning_content")
-            if rc and show_reasoning:
-                if not printed_reasoning:
+            # Parse delta from the chunk
+            choices = parsed_chunk.get("choices", [])
+            if not choices:
+                continue
+
+            delta = choices[0].get("delta", {})
+            reasoning_token = delta.get("reasoning_content", "")
+            content_token = delta.get("content", "")
+
+            # Print reasoning token if applicable
+            if show_reasoning and reasoning_token:
+                if not reasoning_started:
                     print("\n🧠 Reasoning: ", end="", flush=True)
-                    printed_reasoning = True
-                print(rc, end="", flush=True)
-                reasoning_content += rc
+                    reasoning_started = True
+                print(f"\033[90m{reasoning_token}\033[0m", end="", flush=True)
+                reasoning_content += reasoning_token
 
-            # content tokens
-            content_part = delta.get("content")
-            if content_part:
-                if not printed_answer:
-                    if show_reasoning and printed_reasoning:
-                        print("\n💬 Response: ", end="", flush=True)
+            # Print content token
+            if content_token:
+                if not content_started:
+                    if show_reasoning and reasoning_started:
+                        print(f"\n💬 Response: ", end="", flush=True)
                     else:
                         print("Assistant: ", end="", flush=True)
-                    printed_answer = True
-                print(content_part, end="", flush=True)
-                full_response += content_part
+                    content_started = True
+                print(content_token, end="", flush=True)
+                full_response += content_token
+
+        print()  # Ensure newline after response
 
-        print()  # newline
         if show_reasoning:
-            if printed_reasoning or printed_answer:
+            if reasoning_started or content_started:
                 print("\nStreaming completed.")
-            if printed_reasoning:
+            if reasoning_started:
                 print(f"Reasoning tokens: {len(reasoning_content.split())}")
-            if printed_answer:
+            if content_started:
                 print(f"Response tokens: {len(full_response.split())}")
-            if finish_reason:
-                print(f"Finish reason: {finish_reason}")
 
         return full_response
-    
-    async def demo_completions(self) -> None:
-        print("=" * 60)
-        print("COMPLETIONS DEMO")
-        print("=" * 60)
 
-        response = await call_completions(
-            client=self.client,
-            model=self.model,
-            prompt=COMPLETIONS_PROMPT,
-            endpoint_name=self.endpoint_name,
-            max_tokens=MAX_TOKENS,
-            temperature=DEFAULT_TEMPERATURE,
-        )
-        print("\nResponse:")
-        print(json.dumps(response, indent=2))
+    def test_tool_support(self) -> bool:
+        """Test if the endpoint supports function calling"""
+        log.debug("Testing endpoint tool calling support...")
 
-    async def demo_chat(self, use_streaming: bool = True) -> None:
-        print("=" * 60)
-        print(f"CHAT COMPLETIONS DEMO {'(STREAMING)' if use_streaming else '(NON-STREAMING)'}")
-        print("=" * 60)
-
-        messages = [{"role": "user", "content": CHAT_PROMPT}]
-
-        if use_streaming:
-            stream = await stream_chat_completions(
-                client=self.client,
-                model=self.model, 
-                messages=messages,
-                endpoint_name=self.endpoint_name,
-                max_tokens=MAX_TOKENS,
-                temperature=DEFAULT_TEMPERATURE
-            )
-            try:
-                await self.handle_streaming_response(stream, show_reasoning=True)
-            except Exception as e:
-                log.error("\nError during streaming: %s", e, exc_info=True)
-        else:
-            response = await call_chat_completions(
-                client=self.client,
-                model=self.model, 
-                messages=messages,
-                endpoint_name=self.endpoint_name,
-                max_tokens=MAX_TOKENS,
-                temperature=DEFAULT_TEMPERATURE
-            )
-            choice = (response.get("choices") or [{}])[0]
-            message = choice.get("message", {})
-            content = message.get("content", "")
-            reasoning = message.get("reasoning_content", "") or message.get("reasoning", "")
-            if reasoning:
-                print(f"\n🧠 Reasoning: \033[90m{reasoning}\033[0m")
-            print(f"\n💬 Assistant: {content}")
-            print(f"\nFull Response:\n{json.dumps(response, indent=2)}")
-
-    async def test_tool_support(self) -> bool:
-        """Probe that tool schema is accepted (no actual call)"""
+        # Try a simple request with minimal tools to test support
         messages = [{"role": "user", "content": "Hello"}]
         minimal_tool = [
             {
@@ -293,158 +277,179 @@ class APIDemo:
                 "function": {"name": "test_function", "description": "Test function"},
             }
         ]
+
+        config = ChatCompletionConfig(
+            model=self.model,
+            messages=messages,
+            max_tokens=10,
+            tools=minimal_tool,
+            tool_choice="none",  # Don't actually call the tool
+        )
+
         try:
-            _ = await call_chat_completions(
-                client=self.client,
-                model=self.model,
-                messages=messages,
-                endpoint_name=self.endpoint_name,
-                tools=minimal_tool,
-                tool_choice="none",
-                max_tokens=10
-            )
+            response = self.client.call_chat_completions(config)
             return True
         except Exception as e:
-            log.error("Endpoint does not support tool calling: %s", e)
+            log.error(f"Error: Endpoint does not support tool calling: {e}")
             return False
 
-    async def demo_ls_tool(self) -> None:
-        """Ask to list files using function calling, then provide final analysis"""
+    def demo_completions(self) -> None:
+        """Demo: test basic completions endpoint"""
+        print("=" * 60)
+        print("COMPLETIONS DEMO")
+        print("=" * 60)
+
+        config = CompletionConfig(
+            model=self.model, prompt=COMPLETIONS_PROMPT, stream=False
+        )
+
+        log.info(
+            f"Testing completions with model '{self.model}' and prompt: '{config.prompt}'"
+        )
+        response = self.client.call_completions(config)
+
+        if isinstance(response, dict):
+            print("\nResponse:")
+            print(json.dumps(response, indent=2))
+        else:
+            log.error("Unexpected response format")
+
+    def demo_chat(self, use_streaming: bool = True) -> None:
+        """
+        Demo: test chat completions endpoint with optional streaming
+        """
+        print("=" * 60)
+        print(
+            f"CHAT COMPLETIONS DEMO {'(STREAMING)' if use_streaming else '(NON-STREAMING)'}"
+        )
+        print("=" * 60)
+
+        config = ChatCompletionConfig(
+            model=self.model,
+            messages=[{"role": "user", "content": CHAT_PROMPT}],
+            stream=use_streaming,
+        )
+
+        log.info(f"Testing chat completions with model '{self.model}'...")
+        response = self.client.call_chat_completions(config)
+
+        if use_streaming:
+            try:
+                self.handle_streaming_response(response, show_reasoning=True)
+            except Exception as e:
+                log.error(f"\nError during streaming: {e}")
+                import traceback
+
+                traceback.print_exc()
+                return
+
+        else:
+            if isinstance(response, dict):
+                choice = response.get("choices", [{}])[0]
+                message = choice.get("message", {})
+                content = message.get("content", "")
+                reasoning = message.get("reasoning_content", "") or message.get(
+                    "reasoning", ""
+                )
+
+                if reasoning:
+                    print(f"\n🧠 Reasoning: \033[90m{reasoning}\033[0m")
+
+                print(f"\n💬 Assistant: {content}")
+                print(f"\nFull Response:")
+                print(json.dumps(response, indent=2))
+            else:
+                log.error("Unexpected response format")
+
+    def demo_ls_tool(self) -> None:
+        """Demo: ask LLM to list files in the current directory and describe what it sees"""
         print("=" * 60)
         print("TOOL USE DEMO: List Directory Contents")
         print("=" * 60)
 
-        if not await self.test_tool_support():
+        # Test if tools are supported first
+        if not self.test_tool_support():
             return
 
-        messages: List[Dict[str, Any]] = [{"role": "user", "content": TOOLS_PROMPT}]
+        # Request with tool available
+        messages = [{"role": "user", "content": TOOLS_PROMPT}]
 
-        # First pass: let the model decide tools, stream tool_calls and partial content
-        stream = await stream_chat_completions(
-            client=self.client,
+        config = ChatCompletionConfig(
             model=self.model,
             messages=messages,
-            endpoint_name=self.endpoint_name,
             tools=self.tool_manager.get_ls_tool_definition(),
             tool_choice="auto",
-            max_tokens=MAX_TOKENS,
-            temperature=DEFAULT_TEMPERATURE,
         )
 
-        assistant_content_buf: List[str] = []
-        tool_calls_state: Dict[int, Dict[str, Any]] = {}
-        printed_reasoning = False
-        printed_answer = False
+        log.info(f"Making initial request with tool using model '{self.model}'...")
+        response = self.client.call_chat_completions(config)
 
-        async for chunk in stream:
-            choice = (chunk.get("choices") or [{}])[0]
-            delta = choice.get("delta", {})
+        if not isinstance(response, dict):
+            raise ValueError("Expected dict response for tool use")
 
-            rc = delta.get("reasoning_content")
-            if rc:
-                if not printed_reasoning:
-                    printed_reasoning = True
-                    print("🧠 Reasoning: ", end="", flush=True)
-                print(rc, end="", flush=True)
+        choice = response.get("choices", [{}])[0]
+        message = choice.get("message", {})
 
-            content_part = delta.get("content")
-            if content_part:
-                assistant_content_buf.append(content_part)
-                if not printed_answer:
-                    printed_answer = True
-                    print("\n💬 Response: ", end="", flush=True)
-                print(content_part, end="", flush=True)
+        print(f"Assistant response: {message.get('content', 'No content')}")
 
-            if "tool_calls" in delta and delta["tool_calls"]:
-                for tc_delta in delta["tool_calls"]:
-                    _merge_tool_call_delta(tool_calls_state, tc_delta)
+        # Check for tool calls
+        tool_calls = message.get("tool_calls")
+        if not tool_calls:
+            raise ValueError(
+                "No tool calls made - model may not support function calling"
+            )
 
-        # If no tool calls, we’re done.
-        if not tool_calls_state:
-            print("\n(No tool calls were made.)")
-            return
+        print(f"Tool calls detected: {len(tool_calls)}")
 
-        # Build assistant message with tool_calls
-        assistant_message = {
-            "role": "assistant",
-            "content": "".join(assistant_content_buf) if assistant_content_buf else None,
-            "tool_calls": _tool_state_to_message_tool_calls(tool_calls_state),
-        }
-        messages.append(assistant_message)
+        # Execute the tool call
+        for tool_call in tool_calls:
+            function_name = tool_call["function"]["name"]
+            print(f"Executing tool: {function_name}")
 
-        # Execute tools and feed results back
-        for tc in assistant_message["tool_calls"]:
-            tool_name = (tc.get("function") or {}).get("name")
-            call_id = tc.get("id")
-            raw_args = (tc.get("function") or {}).get("arguments") or "{}"
+            tool_result = self.tool_manager.execute_tool_call(tool_call)
+            print(f"Tool result:\n{tool_result}")
 
-            try:
-                args = json.loads(raw_args) if raw_args.strip() else {}
-            except Exception as e:
-                tool_result = json.dumps({"error": f"Argument parse failed: {str(e)}", "raw_arguments": raw_args})
-                messages.append({"role": "tool", "tool_call_id": call_id, "content": tool_result})
-                continue
+            # Add tool result and continue conversation
+            messages.append(message)  # Add assistant's message with tool call
+            messages.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": tool_call["id"],
+                    "content": tool_result,
+                }
+            )
 
-            try:
-                if tool_name == "list_files":
-                    tool_result = self.tool_manager.list_files()
-                else:
-                    tool_result = json.dumps({"error": f"Unknown tool '{tool_name}'"})
-            except Exception as e:
-                tool_result = json.dumps({"error": f"Tool '{tool_name}' failed: {str(e)}"})
+            # Get final response
+            final_config = ChatCompletionConfig(
+                model=self.model,
+                messages=messages,
+                tools=self.tool_manager.get_ls_tool_definition(),
+            )
 
-            print("\n[Tool executed]", tool_name)
-            print(tool_result[:500] + ("..." if len(tool_result) > 500 else ""))
-            messages.append({"role": "tool", "tool_call_id": call_id, "content": tool_result})
+            print("Getting final response...")
+            final_response = self.client.call_chat_completions(final_config)
 
-        # Second pass: get final streamed answer after tool results
-        stream2 = await stream_chat_completions(
-            client=self.client,
-            model=self.model,
-            messages=messages,
-            endpoint_name=self.endpoint_name,
-            max_tokens=MAX_TOKENS,
-            temperature=DEFAULT_TEMPERATURE,
-        )
+            if isinstance(final_response, dict):
+                final_choice = final_response.get("choices", [{}])[0]
+                final_message = final_choice.get("message", {})
+                final_content = final_message.get("content", "")
 
-        final_buf = []
-        printed_reasoning2 = False
-        printed_answer2 = False
+                print("\n" + "=" * 60)
+                print("FINAL LLM ANALYSIS:")
+                print("=" * 60)
+                print(final_content)
+                print("=" * 60)
 
-        async for chunk in stream2:
-            choice = (chunk.get("choices") or [{}])[0]
-            delta = choice.get("delta", {})
-
-            rc2 = delta.get("reasoning_content")
-            if rc2:
-                if not printed_reasoning2:
-                    printed_reasoning2 = True
-                    print("\n🧠 Reasoning (post-tools): ", end="", flush=True)
-                print(rc2, end="", flush=True)
-
-            c2 = delta.get("content")
-            if c2:
-                final_buf.append(c2)
-                if not printed_answer2:
-                    printed_answer2 = True
-                    print("\n💬 Response (final): ", end="", flush=True)
-                print(c2, end="", flush=True)
-
-        print("\n" + "=" * 60)
-        print("FINAL LLM ANALYSIS:")
-        print("=" * 60)
-        print("".join(final_buf))
-        print("=" * 60)
-
-    async def interactive_chat(self) -> None:
+    def interactive_chat(self) -> None:
         """Interactive chat session with streaming"""
         print("=" * 60)
         print("INTERACTIVE STREAMING CHAT")
         print("=" * 60)
+        print(f"Using model: {self.model}")
         print("Type 'quit' to exit, 'clear' to clear history")
         print()
 
-        messages: List[Dict[str, Any]] = []
+        messages = []
 
         while True:
             try:
@@ -462,16 +467,16 @@ class APIDemo:
 
                 messages.append({"role": "user", "content": user_input})
 
-                print("Assistant: ", end="", flush=True)
-                stream = await stream_chat_completions(
-                    client=self.client,
-                    model=self.model, 
-                    messages=messages,
-                    endpoint_name=self.endpoint_name,
-                    max_tokens=MAX_TOKENS, 
-                    temperature=0.7
+                config = ChatCompletionConfig(
+                    model=self.model, messages=messages, stream=True, temperature=0.7
+                )
+
+                print("Assistant: ", end="", flush=True)
+
+                response = self.client.call_chat_completions(config)
+                assistant_content = self.handle_streaming_response(
+                    response, show_reasoning=True
                 )
-                assistant_content = await self.handle_streaming_response(stream, show_reasoning=True)
 
                 # Add assistant response to conversation history
                 messages.append({"role": "assistant", "content": assistant_content})
@@ -480,66 +485,115 @@ class APIDemo:
                 print("\n👋 Chat interrupted. Goodbye!")
                 break
             except Exception as e:
-                log.error("\nError: %s", e)
+                log.error(f"\nError: {e}")
                 continue
 
 
-# ---------------------- CLI ----------------------
-def build_arg_parser() -> argparse.ArgumentParser:
-    p = argparse.ArgumentParser(description="Vast vLLM Demo (Serverless SDK)")
-    p.add_argument("--model", default=DEFAULT_MODEL, help=f"Model to use for requests (default: {DEFAULT_MODEL})")
-    p.add_argument("--endpoint", default=ENDPOINT_NAME, help=f"Vast endpoint name (default: {ENDPOINT_NAME})")
+def main():
+    """Main function with CLI switches for different tests"""
+    from lib.test_utils import test_args
 
-    modes = p.add_mutually_exclusive_group(required=False)
-    modes.add_argument("--completion", action="store_true", help="Test completions endpoint")
-    modes.add_argument("--chat", action="store_true", help="Test chat completions endpoint (non-streaming)")
-    modes.add_argument("--chat-stream", action="store_true", help="Test chat completions endpoint with streaming")
-    modes.add_argument("--tools", action="store_true", help="Test function calling with ls tool (non-streaming+streamed phases)")
-    modes.add_argument("--interactive", action="store_true", help="Start interactive streaming chat session")
-    return p
+    # Add mandatory model argument
+    test_args.add_argument(
+        "--model", required=True, help="Model to use for requests (required)"
+    )
 
+    # Add test mode arguments
+    test_args.add_argument(
+        "--completion", action="store_true", help="Test completions endpoint"
+    )
+    test_args.add_argument(
+        "--chat",
+        action="store_true",
+        help="Test chat completions endpoint (non-streaming)",
+    )
+    test_args.add_argument(
+        "--chat-stream",
+        action="store_true",
+        help="Test chat completions endpoint with streaming",
+    )
+    test_args.add_argument(
+        "--tools",
+        action="store_true",
+        help="Test function calling with ls tool (non-streaming)",
+    )
+    test_args.add_argument(
+        "--interactive",
+        action="store_true",
+        help="Start interactive streaming chat session",
+    )
 
-async def main_async():
-    args = build_arg_parser().parse_args()
+    args = test_args.parse_args()
 
-    selected = sum([args.completion, args.chat, args.chat_stream, args.tools, args.interactive])
-    if selected == 0:
+    # Check that only one test mode is selected
+    test_modes = [
+        args.completion,
+        args.chat,
+        args.chat_stream,
+        args.tools,
+        args.interactive,
+    ]
+    selected_count = sum(test_modes)
+
+    if selected_count == 0:
         print("Please specify exactly one test mode:")
         print("  --completion    : Test completions endpoint")
         print("  --chat          : Test chat completions endpoint (non-streaming)")
         print("  --chat-stream   : Test chat completions endpoint with streaming")
-        print("  --tools         : Test function calling with ls tool")
+        print("  --tools         : Test function calling with ls tool (non-streaming)")
         print("  --interactive   : Start interactive streaming chat session")
-        print(f"\nExample: python {os.path.basename(sys.argv[0])} --model Qwen/Qwen3-8B --chat-stream --endpoint my-vllm-endpoint")
+        print(
+            f"\nExample: python {sys.argv[0]} --model Qwen/Qwen3-8B --chat-stream -k YOUR_KEY -e YOUR_ENDPOINT"
+        )
         sys.exit(1)
-    elif selected > 1:
+    elif selected_count > 1:
         print("Please specify exactly one test mode")
         sys.exit(1)
 
-    print("=" * 60)
-    print(f"Using model: {args.model}")
-    print(f"Using endpoint: {args.endpoint}")
-
-
     try:
-        async with Serverless() as client:
-            demo = APIDemo(client, args.model, args.endpoint, ToolManager())
+        endpoint_api_key = Endpoint.get_endpoint_api_key(
+            endpoint_name=args.endpoint_group_name,
+            account_api_key=args.api_key,
+            instance=args.instance,
+        )
 
-            if args.completion:
-                await demo.demo_completions()
-            elif args.chat:
-                await demo.demo_chat(use_streaming=False)
-            elif args.chat_stream:
-                await demo.demo_chat(use_streaming=True)
-            elif args.tools:
-                await demo.demo_ls_tool()
-            elif args.interactive:
-                await demo.interactive_chat()
+        if not endpoint_api_key:
+            log.error(
+                f"Could not retrieve API key for endpoint '{args.endpoint_group_name}'. Exiting."
+            )
+            sys.exit(1)
+
+        # Create the core API client
+        client = APIClient(
+            endpoint_group_name=args.endpoint_group_name,
+            api_key=args.api_key,
+            server_url=Endpoint.get_autoscaler_server_url(args.instance),
+            endpoint_api_key=endpoint_api_key,
+        )
+
+        # Create tool manager and demo (passing the model parameter)
+        tool_manager = ToolManager()
+        demo = APIDemo(client, args.model, tool_manager)
+
+        print(f"Using model: {args.model}")
+        print("=" * 60)
+
+        # Run the selected test
+        if args.completion:
+            demo.demo_completions()
+        elif args.chat:
+            demo.demo_chat(use_streaming=False)
+        elif args.chat_stream:
+            demo.demo_chat(use_streaming=True)
+        elif args.tools:
+            demo.demo_ls_tool()
+        elif args.interactive:
+            demo.interactive_chat()
 
     except Exception as e:
-        log.error("Error during test: %s", e, exc_info=True)
+        log.error(f"Error during test: {e}", exc_info=True)
         sys.exit(1)
 
 
 if __name__ == "__main__":
-    asyncio.run(main_async())
+    main()

workers/openai/server.py

@@ -11,7 +11,6 @@ MODEL_SERVER_START_LOG_MSG = [
     "llama runner started",  # Ollama
     '"message":"Connected","target":"text_generation_router"',  # TGI
     '"message":"Connected","target":"text_generation_router::server"',  # TGI
-    "main: model loaded" # llama.cpp
 ]
 
 MODEL_SERVER_ERROR_LOG_MSGS = [
@@ -35,7 +34,6 @@ backend = Backend(
     model_server_url=os.environ["MODEL_SERVER_URL"],
     model_log_file=os.environ["MODEL_LOG"],
     allow_parallel_requests=True,
-    max_wait_time=600.0,
     benchmark_handler=CompletionsHandler(benchmark_runs=3, benchmark_words=256),
     log_actions=[
         *[(LogAction.ModelLoaded, info_msg) for info_msg in MODEL_SERVER_START_LOG_MSG],

workers/tgi/README.md

@@ -1,103 +1,19 @@
-# HuggingFace TGI PyWorker
+This is the base PyWorker for TGI, designed to create PyWorkers that can utilize various LLMs. It offers two primary endpoints:
 
-This is the base PyWorker for HuggingFace Text Generation Inference (TGI) servers. See the [Serverless documentation](https://docs.vast.ai/serverless) for guides and how-to's.
+1. `generate`: Generates the LLM's response to a given prompt in a single request.
+2. `generate_stream`: Streams the LLM's response token by token.
 
-## Instance Setup
-
-1. Pick a template
-
-This worker is compatible with any TGI backend. We have a template you can use or you can create your own.
-
-- [HuggingFace TGI](https://cloud.vast.ai/?ref_id=62897&creator_id=62897&name=TGI%20(Serverless))
-
-The template can be configured via the template interface. You may want to change the model or startup arguments.
-
-2. Follow the [getting started guide](https://docs.vast.ai/documentation/serverless/quickstart) for help with configuring your serverless setup. For testing, we recommend that you use the default options presented by the web interface.
-
-## Client Setup (Demo)
-
-1. Clone the PyWorker repository to your local machine and install the necessary requirements for running the test client.
-
-```bash
-git clone https://github.com/vast-ai/pyworker
-cd pyworker
-pip install uv
-uv venv -p 3.12
-source .venv/bin/activate
-uv pip install -r requirements.txt
-```
-
-## Using the Test Client
-
-The test client demonstrates both streaming and non-streaming generation using TGI's native API.
-
-First, set your API key as an environment variable:
-
-```bash
-export VAST_API_KEY=<your_api_key>
-```
-
-The `--endpoint` flag is optional. If not provided, it defaults to `my-tgi-endpoint`.
-
-### Generate (Streaming)
-
-Call to `/generate_stream` with streaming response:
-
-```bash
-python -m workers.tgi.client --generate-stream --endpoint <ENDPOINT_NAME>
-```
-
-### Generate (Non-Streaming)
-
-Call to `/generate` with json response:
-
-```bash
-python -m workers.tgi.client --generate --endpoint <ENDPOINT_NAME>
-```
-
-### Interactive Session (Streaming)
-
-Interactive session with streaming responses. Type `quit` to exit.
-
-```bash
-python -m workers.tgi.client --interactive --endpoint <ENDPOINT_NAME>
-```
-
-## API Endpoints
-
-TGI provides two primary endpoints:
-
-### Generate (Non-Streaming)
-
-`/generate` - Returns the complete response in a single request.
+Both endpoints use the following API payload format:
 
 ```json
 {
-  "inputs": "Your prompt here",
+  "inputs": "PROMPT",
   "parameters": {
-    "max_new_tokens": 1024,
-    "temperature": 0.7,
-    "return_full_text": false
+    "max_new_tokens": 250
   }
 }
 ```
 
-### Generate Stream (Streaming)
-
-`/generate_stream` - Streams the response token by token.
-
-```json
-{
-  "inputs": "Your prompt here",
-  "parameters": {
-    "max_new_tokens": 1024,
-    "temperature": 0.7,
-    "do_sample": true,
-    "return_full_text": false
-  }
-}
-```
-
-## Performance Notes
-
-The `max_new_tokens` parameter (not the prompt size) primarily impacts performance. For example, if an instance is benchmarked to process 100 tokens per second, a request with `max_new_tokens = 200` will take approximately 2 seconds to complete.
+Note that the max_new_tokens parameter, rather than the prompt size, impacts performance. For example, if an
+instance is benchmarked to process 100 tokens per second, a request with max_new_tokens = 200 will take
+approximately 2 seconds to complete.

workers/tgi/client.py

@@ -1,13 +1,11 @@
 import logging
-import json
-import os
 import sys
-import argparse
+import json
+from urllib.parse import urljoin
+import requests
+from utils.endpoint_util import Endpoint
+from utils.ssl import get_cert_file_path
 
-from vastai import Serverless
-import asyncio
-
-# ---------------------- Logging ----------------------
 logging.basicConfig(
     level=logging.DEBUG,
     format="%(asctime)s[%(levelname)-5s] %(message)s",
@@ -15,208 +13,113 @@ logging.basicConfig(
 )
 log = logging.getLogger(__file__)
 
-# ---------------------- Defaults ----------------------
-DEFAULT_PROMPT = "Think step by step: Tell me about the Python programming language."
 
-ENDPOINT_NAME = "TGI-Prod2"       # change this to your TGI endpoint name
-MAX_TOKENS = 1024
-DEFAULT_TEMPERATURE = 0.7
-
-
-# ---------------------- API Calls ----------------------
-async def call_generate(client: Serverless, *, endpoint_name: str, prompt: str, **kwargs) -> dict:
-    """Non-streaming generation via /generate endpoint"""
-    endpoint = await client.get_endpoint(name=endpoint_name)
-
-    payload = {
-        "inputs": prompt,
-        "parameters": {
-            "max_new_tokens": kwargs.get("max_tokens", MAX_TOKENS),
-            "temperature": kwargs.get("temperature", DEFAULT_TEMPERATURE),
-            "return_full_text": False,
-        }
+def call_generate(endpoint_group_name: str, api_key: str, server_url: str) -> None:
+    WORKER_ENDPOINT = "/generate"
+    COST = 100
+    route_payload = {
+        "endpoint": endpoint_group_name,
+        "api_key": api_key,
+        "cost": COST,
     }
-    log.debug("POST /generate %s", json.dumps(payload)[:500])
-    resp = await endpoint.request("/generate", payload, cost=payload["parameters"]["max_new_tokens"])
-    return resp["response"]
-
-
-async def call_generate_stream(client: Serverless, *, endpoint_name: str, prompt: str, **kwargs):
-    """Streaming generation via /generate_stream endpoint"""
-    endpoint = await client.get_endpoint(name=endpoint_name)
-
-    payload = {
-        "inputs": prompt,
-        "parameters": {
-            "max_new_tokens": kwargs.get("max_tokens", MAX_TOKENS),
-            "temperature": kwargs.get("temperature", DEFAULT_TEMPERATURE),
-            "do_sample": True,
-            "return_full_text": False,
-        }
-    }
-    log.debug("STREAM /generate_stream %s", json.dumps(payload)[:500])
-    resp = await endpoint.request(
-        "/generate_stream",
-        payload,
-        cost=payload["parameters"]["max_new_tokens"],
-        stream=True,
+    response = requests.post(
+        urljoin(server_url, "/route/"),
+        json=route_payload,
+        timeout=4,
     )
-    return resp["response"]  # async generator
+    response.raise_for_status()  # Raise an exception for bad status codes
+    message = response.json()
+    url = message["url"]
+
+    auth_data = dict(
+        signature=message["signature"],
+        cost=message["cost"],
+        endpoint=message["endpoint"],
+        reqnum=message["reqnum"],
+        url=url,
+    )
+
+    payload = dict(inputs="tell me about cats", parameters=dict(max_new_tokens=500))
+    req_data = dict(payload=payload, auth_data=auth_data)
+    url = urljoin(url, WORKER_ENDPOINT)
+    print(f"url: {url}")
+    response = requests.post(
+        url,
+        json=req_data,
+        verify=get_cert_file_path(),
+    )
+    response.raise_for_status()
+    res = response.json()
+    print(res)
 
 
-# ---------------------- Demo Runner ----------------------
-class APIDemo:
-    """Demo and testing functionality for the TGI API client"""
-
-    def __init__(self, client: Serverless, endpoint_name: str):
-        self.client = client
-        self.endpoint_name = endpoint_name
-
-    async def handle_streaming_response(self, stream) -> str:
-        """Process streaming response and print tokens"""
-        full_response = ""
-        printed_answer = False
-
-        async for event in stream:
-            tok = (event.get("token") or {}).get("text")
-            if tok:
-                if not printed_answer:
-                    printed_answer = True
-                    print("\n💬 Response: ", end="", flush=True)
-                print(tok, end="", flush=True)
-                full_response += tok
-
-        print()  # newline
-        if printed_answer:
-            print(f"\nStreaming completed. Response tokens: {len(full_response.split())}")
-
-        return full_response
-
-    async def demo_generate(self) -> None:
-        """Demo non-streaming generation"""
-        print("=" * 60)
-        print("GENERATE DEMO (NON-STREAMING)")
-        print("=" * 60)
-
-        response = await call_generate(
-            client=self.client,
-            endpoint_name=self.endpoint_name,
-            prompt=DEFAULT_PROMPT,
-            max_tokens=MAX_TOKENS,
-            temperature=DEFAULT_TEMPERATURE,
-        )
-        
-        print(f"\n💬 Response: {response.get('generated_text', '')}")
-        print(f"\nFull Response:\n{json.dumps(response, indent=2)}")
-
-    async def demo_generate_stream(self) -> None:
-        """Demo streaming generation"""
-        print("=" * 60)
-        print("GENERATE DEMO (STREAMING)")
-        print("=" * 60)
-
-        stream = await call_generate_stream(
-            client=self.client,
-            endpoint_name=self.endpoint_name,
-            prompt=DEFAULT_PROMPT,
-            max_tokens=MAX_TOKENS,
-            temperature=DEFAULT_TEMPERATURE,
-        )
-
-        try:
-            await self.handle_streaming_response(stream)
-        except Exception as e:
-            log.error("\nError during streaming: %s", e, exc_info=True)
-
-    async def interactive_chat(self) -> None:
-        """Interactive session with streaming generation"""
-        print("=" * 60)
-        print("INTERACTIVE STREAMING SESSION")
-        print("=" * 60)
-        print(f"Using endpoint: {self.endpoint_name}")
-        print("Type 'quit' to exit")
-        print()
-
-        while True:
+def call_generate_stream(
+    endpoint_group_name: str, api_key: str, server_url: str
+) -> None:
+    WORKER_ENDPOINT = "/generate_stream"
+    COST = 100
+    route_payload = {
+        "endpoint": endpoint_group_name,
+        "api_key": api_key,
+        "cost": COST,
+    }
+    response = requests.post(
+        urljoin(server_url, "/route/"),
+        json=route_payload,
+        timeout=4,
+    )
+    response.raise_for_status()  # Raise an exception for bad status codes
+    message = response.json()
+    url = message["url"]
+    print(f"url: {url}")
+    auth_data = dict(
+        signature=message["signature"],
+        cost=message["cost"],
+        endpoint=message["endpoint"],
+        reqnum=message["reqnum"],
+        url=message["url"],
+    )
+    payload = dict(inputs="tell me about dogs", parameters=dict(max_new_tokens=500))
+    req_data = dict(payload=payload, auth_data=auth_data)
+    url = urljoin(url, WORKER_ENDPOINT)
+    response = requests.post(url, json=req_data, stream=True)
+    response.raise_for_status()  # Raise an exception for bad status codes
+    for line in response.iter_lines():
+        payload = line.decode().lstrip("data:").rstrip()
+        if payload:
             try:
-                user_input = input("You: ").strip()
-
-                if user_input.lower() == "quit":
-                    print("👋 Goodbye!")
-                    break
-                elif not user_input:
-                    continue
-
-                print("Assistant: ", end="", flush=True)
-                stream = await call_generate_stream(
-                    client=self.client,
-                    endpoint_name=self.endpoint_name,
-                    prompt=user_input,
-                    max_tokens=MAX_TOKENS,
-                    temperature=DEFAULT_TEMPERATURE,
-                )
-
-                full_response = ""
-                async for event in stream:
-                    tok = (event.get("token") or {}).get("text")
-                    if tok:
-                        print(tok, end="", flush=True)
-                        full_response += tok
-                print()  # newline
-
-            except KeyboardInterrupt:
-                print("\n👋 Session interrupted. Goodbye!")
-                break
-            except Exception as e:
-                log.error("\nError: %s", e)
+                data = json.loads(payload)
+                print(data["token"]["text"], end="")
+                sys.stdout.flush()
+            except (json.JSONDecodeError, KeyError) as e:
+                log.warning(f"Failed to parse streaming response: {e}")
                 continue
-
-
-# ---------------------- CLI ----------------------
-def build_arg_parser() -> argparse.ArgumentParser:
-    p = argparse.ArgumentParser(description="Vast TGI Demo (Serverless SDK)")
-    p.add_argument("--endpoint", default=ENDPOINT_NAME, help=f"Vast endpoint name (default: {ENDPOINT_NAME})")
-
-    modes = p.add_mutually_exclusive_group(required=False)
-    modes.add_argument("--generate", action="store_true", help="Test generate endpoint (non-streaming)")
-    modes.add_argument("--generate-stream", action="store_true", help="Test generate endpoint with streaming")
-    modes.add_argument("--interactive", action="store_true", help="Start interactive streaming session")
-    return p
-
-
-async def main_async():
-    args = build_arg_parser().parse_args()
-
-    selected = sum([args.generate, args.generate_stream, args.interactive])
-    if selected == 0:
-        print("Please specify exactly one test mode:")
-        print("  --generate        : Test generate endpoint (non-streaming)")
-        print("  --generate-stream : Test generate endpoint with streaming")
-        print("  --interactive     : Start interactive streaming session")
-        print(f"\nExample: python {os.path.basename(sys.argv[0])} --generate-stream --endpoint my-tgi-endpoint")
-        sys.exit(1)
-    elif selected > 1:
-        print("Please specify exactly one test mode")
-        sys.exit(1)
-
-    print("=" * 60)
-    print(f"Using endpoint: {args.endpoint}")
-
-    try:
-        async with Serverless() as client:
-            demo = APIDemo(client, args.endpoint)
-
-            if args.generate:
-                await demo.demo_generate()
-            elif args.generate_stream:
-                await demo.demo_generate_stream()
-            elif args.interactive:
-                await demo.interactive_chat()
-
-    except Exception as e:
-        log.error("Error during test: %s", e, exc_info=True)
-        sys.exit(1)
+    print()
 
 
 if __name__ == "__main__":
-    asyncio.run(main_async())
+    from lib.test_utils import test_args
+
+    args = test_args.parse_args()
+
+    endpoint_api_key = Endpoint.get_endpoint_api_key(
+        endpoint_name=args.endpoint_group_name,
+        account_api_key=args.api_key,
+        instance=args.instance,
+    )
+    if endpoint_api_key:
+        try:
+            call_generate(
+                api_key=endpoint_api_key,
+                endpoint_group_name=args.endpoint_group_name,
+                server_url=args.server_url,
+            )
+            call_generate_stream(
+                api_key=endpoint_api_key,
+                endpoint_group_name=args.endpoint_group_name,
+                server_url=args.server_url,
+            )
+        except Exception as e:
+            log.error(f"Error during API call: {e}")
+    else:
+        log.error(f"Failed to get API key for endpoint {args.endpoint_group_name} ")