custom hf jobs tool impl

agent/config_local_mcp.json

@@ -0,0 +1,16 @@
+{
+  "model_name": "anthropic/claude-sonnet-4-5-20250929",
+  "tools": [],
+  "system_prompt_path": "",
+  "mcpServers": {
+    "hf-mcp-server": {
+      "command": "node",
+      "args": [
+        "/Users/akseljoonas/Documents/hf-agent/hf-mcp-server/packages/app/dist/server/stdio.js"
+      ],
+      "env": {
+        "DEFAULT_HF_TOKEN": "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"
+      }
+    }
+  }
+}

agent/core/tools.py

@@ -13,6 +13,7 @@ from lmnr import observe
 from mcp.types import EmbeddedResource, ImageContent, TextContent
 
 from agent.config import MCPServerConfig
+from agent.tools.hf.jobs import HF_JOBS_TOOL_SPEC, hf_jobs_handler
 
 # Suppress aiohttp deprecation warning
 warnings.filterwarnings(
@@ -262,4 +263,10 @@ def create_builtin_tools() -> list[ToolSpec]:
             },
             handler=write_file_handler,
         ),
+        ToolSpec(
+            name=HF_JOBS_TOOL_SPEC["name"],
+            description=HF_JOBS_TOOL_SPEC["description"],
+            parameters=HF_JOBS_TOOL_SPEC["parameters"],
+            handler=hf_jobs_handler,
+        ),
     ]

agent/tools/hf/__init__.py

@@ -0,0 +1,7 @@
+"""
+Hugging Face tools for the agent
+"""
+from agent.tools.hf.types import ToolResult
+from agent.tools.hf.base import HfApiCall, HfApiError
+
+__all__ = ['ToolResult', 'HfApiCall', 'HfApiError']

agent/tools/hf/base.py

@@ -0,0 +1,84 @@
+"""
+Base API client for Hugging Face API
+
+Ported from: hf-mcp-server/packages/mcp/src/hf-api-call.ts
+"""
+import os
+from typing import Optional, Dict, Any, TypeVar, Generic
+import httpx
+
+
+TResponse = TypeVar('TResponse')
+
+
+class HfApiError(Exception):
+    """Error from Hugging Face API"""
+
+    def __init__(
+        self,
+        message: str,
+        status: Optional[int] = None,
+        status_text: Optional[str] = None,
+        response_body: Optional[str] = None
+    ):
+        super().__init__(message)
+        self.message = message
+        self.status = status
+        self.status_text = status_text
+        self.response_body = response_body
+
+
+class HfApiCall(Generic[TResponse]):
+    """Base class for making authenticated API calls to Hugging Face"""
+
+    def __init__(
+        self,
+        api_url: str,
+        hf_token: Optional[str] = None,
+        api_timeout: Optional[float] = None
+    ):
+        self.api_url = api_url
+        self.hf_token = hf_token or os.getenv('HF_TOKEN')
+        self.api_timeout = api_timeout or float(os.getenv('HF_API_TIMEOUT', '12.5'))
+
+    async def fetch_from_api(
+        self,
+        url: str,
+        method: str = "GET",
+        json: Optional[Dict[str, Any]] = None,
+        **kwargs
+    ) -> Optional[Any]:
+        """Fetch data from API with auth and error handling"""
+        headers = {
+            "Accept": "application/json",
+            **kwargs.pop("headers", {})
+        }
+
+        if self.hf_token:
+            headers["Authorization"] = f"Bearer {self.hf_token}"
+
+        async with httpx.AsyncClient(timeout=self.api_timeout) as client:
+            if method == "GET":
+                response = await client.get(url, headers=headers, **kwargs)
+            elif method == "POST":
+                response = await client.post(url, headers=headers, json=json, **kwargs)
+            elif method == "DELETE":
+                response = await client.delete(url, headers=headers, **kwargs)
+            elif method == "PUT":
+                response = await client.put(url, headers=headers, json=json, **kwargs)
+            else:
+                raise ValueError(f"Unsupported HTTP method: {method}")
+
+            if not response.is_success:
+                raise HfApiError(
+                    message=f"API request failed: {response.status_code} {response.reason_phrase}",
+                    status=response.status_code,
+                    status_text=response.reason_phrase,
+                    response_body=response.text
+                )
+
+            # Handle empty responses (DELETE often returns empty)
+            if not response.text:
+                return None
+
+            return response.json()

agent/tools/hf/jobs/__init__.py

@@ -0,0 +1,6 @@
+"""
+Hugging Face Jobs tool
+"""
+from agent.tools.hf.jobs.jobs_tool import HfJobsTool, HF_JOBS_TOOL_SPEC, hf_jobs_handler
+
+__all__ = ['HfJobsTool', 'HF_JOBS_TOOL_SPEC', 'hf_jobs_handler']

agent/tools/hf/jobs/api_client.py

@@ -0,0 +1,156 @@
+"""
+Jobs API Client
+
+Ported from: hf-mcp-server/packages/mcp/src/jobs/api-client.ts
+"""
+from typing import Optional, Dict, Any, List
+from agent.tools.hf.base import HfApiCall
+
+
+class JobsApiClient(HfApiCall):
+    """API client for HuggingFace Jobs API"""
+
+    def __init__(self, hf_token: Optional[str] = None, namespace: Optional[str] = None):
+        super().__init__('https://huggingface.co/api', hf_token)
+        self.namespace_cache = namespace
+
+    async def get_namespace(self, namespace: Optional[str] = None) -> str:
+        """
+        Get the namespace (username or org) for the current user
+        Uses cached value or /api/whoami-v2 endpoint as fallback
+        """
+        if namespace:
+            return namespace
+
+        if self.namespace_cache:
+            return self.namespace_cache
+
+        # Fetch from whoami endpoint
+        whoami = await self.fetch_from_api('https://huggingface.co/api/whoami-v2')
+        self.namespace_cache = whoami['name']
+        return self.namespace_cache
+
+    async def run_job(self, job_spec: Dict[str, Any], namespace: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Run a job
+        POST /api/jobs/{namespace}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/jobs/{ns}'
+
+        result = await self.fetch_from_api(url, method='POST', json=job_spec)
+        return result
+
+    async def list_jobs(self, namespace: Optional[str] = None) -> List[Dict[str, Any]]:
+        """
+        List all jobs for a namespace
+        GET /api/jobs/{namespace}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/jobs/{ns}'
+
+        return await self.fetch_from_api(url)
+
+    async def get_job(self, job_id: str, namespace: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Get detailed information about a specific job
+        GET /api/jobs/{namespace}/{jobId}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/jobs/{ns}/{job_id}'
+
+        return await self.fetch_from_api(url)
+
+    async def cancel_job(self, job_id: str, namespace: Optional[str] = None) -> None:
+        """
+        Cancel a running job
+        POST /api/jobs/{namespace}/{jobId}/cancel
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/jobs/{ns}/{job_id}/cancel'
+
+        await self.fetch_from_api(url, method='POST')
+
+    def get_logs_url(self, job_id: str, namespace: str) -> str:
+        """Get logs URL for a job (for SSE streaming)"""
+        return f'https://huggingface.co/api/jobs/{namespace}/{job_id}/logs'
+
+    async def create_scheduled_job(
+        self,
+        spec: Dict[str, Any],
+        namespace: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Create a scheduled job
+        POST /api/scheduled-jobs/{namespace}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/scheduled-jobs/{ns}'
+
+        return await self.fetch_from_api(url, method='POST', json=spec)
+
+    async def list_scheduled_jobs(self, namespace: Optional[str] = None) -> List[Dict[str, Any]]:
+        """
+        List all scheduled jobs
+        GET /api/scheduled-jobs/{namespace}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/scheduled-jobs/{ns}'
+
+        return await self.fetch_from_api(url)
+
+    async def get_scheduled_job(
+        self,
+        scheduled_job_id: str,
+        namespace: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Get details of a scheduled job
+        GET /api/scheduled-jobs/{namespace}/{scheduledJobId}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/scheduled-jobs/{ns}/{scheduled_job_id}'
+
+        return await self.fetch_from_api(url)
+
+    async def delete_scheduled_job(
+        self,
+        scheduled_job_id: str,
+        namespace: Optional[str] = None
+    ) -> None:
+        """
+        Delete a scheduled job
+        DELETE /api/scheduled-jobs/{namespace}/{scheduledJobId}
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/scheduled-jobs/{ns}/{scheduled_job_id}'
+
+        await self.fetch_from_api(url, method='DELETE')
+
+    async def suspend_scheduled_job(
+        self,
+        scheduled_job_id: str,
+        namespace: Optional[str] = None
+    ) -> None:
+        """
+        Suspend a scheduled job
+        POST /api/scheduled-jobs/{namespace}/{scheduledJobId}/suspend
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/scheduled-jobs/{ns}/{scheduled_job_id}/suspend'
+
+        await self.fetch_from_api(url, method='POST')
+
+    async def resume_scheduled_job(
+        self,
+        scheduled_job_id: str,
+        namespace: Optional[str] = None
+    ) -> None:
+        """
+        Resume a suspended scheduled job
+        POST /api/scheduled-jobs/{namespace}/{scheduledJobId}/resume
+        """
+        ns = await self.get_namespace(namespace)
+        url = f'https://huggingface.co/api/scheduled-jobs/{ns}/{scheduled_job_id}/resume'
+
+        await self.fetch_from_api(url, method='POST')

agent/tools/hf/jobs/job_utils.py

@@ -0,0 +1,130 @@
+"""
+Job utility functions
+
+Ported from: hf-mcp-server/packages/mcp/src/jobs/commands/utils.ts
+"""
+import re
+import shlex
+from typing import Dict, Optional, Any, List, Union
+
+
+def parse_timeout(timeout: str) -> int:
+    """Parse timeout string (e.g., "5m", "2h", "30s") to seconds"""
+    time_units = {
+        's': 1,
+        'm': 60,
+        'h': 3600,
+        'd': 86400,
+    }
+
+    match = re.match(r'^(\d+(?:\.\d+)?)(s|m|h|d)$', timeout)
+    if match:
+        value = float(match.group(1))
+        unit = match.group(2)
+        return int(value * time_units[unit])
+
+    # Try to parse as plain number (seconds)
+    try:
+        return int(timeout)
+    except ValueError:
+        raise ValueError(
+            f'Invalid timeout format: "{timeout}". Use format like "5m", "2h", "30s", or plain seconds.'
+        )
+
+
+def parse_image_source(image: str) -> Dict[str, Optional[str]]:
+    """
+    Detect if image is a Space URL and extract spaceId
+    Returns {'dockerImage': ...} or {'spaceId': ...}
+    """
+    space_prefixes = [
+        'https://huggingface.co/spaces/',
+        'https://hf.co/spaces/',
+        'huggingface.co/spaces/',
+        'hf.co/spaces/',
+    ]
+
+    for prefix in space_prefixes:
+        if image.startswith(prefix):
+            return {'dockerImage': None, 'spaceId': image[len(prefix):]}
+
+    # Not a space, treat as docker image
+    return {'dockerImage': image, 'spaceId': None}
+
+
+def parse_command(command: Union[str, List[str]]) -> Dict[str, Any]:
+    """
+    Parse command string or array into command array
+    Uses shlex for POSIX-compliant parsing
+    """
+    # If already an array, return as-is
+    if isinstance(command, list):
+        return {'command': command, 'arguments': []}
+
+    # Parse the command string using shlex for POSIX-compliant parsing
+    try:
+        string_args = shlex.split(command)
+    except ValueError as e:
+        raise ValueError(
+            f'Unsupported shell syntax in command: "{command}". '
+            f'Please use an array format for commands with complex shell operators. Error: {e}'
+        )
+
+    if len(string_args) == 0:
+        raise ValueError(f'Invalid command: "{command}". Command cannot be empty.')
+
+    return {'command': string_args, 'arguments': []}
+
+
+def replace_token_placeholder(value: str, hf_token: Optional[str]) -> str:
+    """Replace HF token placeholder with actual token if available"""
+    if not hf_token:
+        return value
+
+    if value in ('$HF_TOKEN', '${HF_TOKEN}'):
+        return hf_token
+
+    return value
+
+
+def transform_env_map(
+    env_map: Optional[Dict[str, str]],
+    hf_token: Optional[str]
+) -> Optional[Dict[str, str]]:
+    """Transform environment map, replacing token placeholders"""
+    if not env_map:
+        return None
+
+    return {
+        key: replace_token_placeholder(value, hf_token)
+        for key, value in env_map.items()
+    }
+
+
+def create_job_spec(args: Dict[str, Any]) -> Dict[str, Any]:
+    """Create a JobSpec from run command arguments"""
+    # Validate required fields
+    if not args.get('image'):
+        raise ValueError('image parameter is required. Provide a Docker image (e.g., "python:3.12") or Space URL.')
+    if not args.get('command'):
+        raise ValueError('command parameter is required. Provide a command as string or array.')
+
+    image_source = parse_image_source(args['image'])
+    command_parsed = parse_command(args['command'])
+    timeout_seconds = parse_timeout(args['timeout']) if args.get('timeout') else None
+    environment = transform_env_map(args.get('env'), args.get('hfToken')) or {}
+    secrets = transform_env_map(args.get('secrets'), args.get('hfToken')) or {}
+
+    spec = {
+        **{k: v for k, v in image_source.items() if v is not None},
+        'command': command_parsed['command'],
+        'arguments': command_parsed['arguments'],
+        'flavor': args.get('flavor', 'cpu-basic'),
+        'environment': environment,
+        'secrets': secrets,
+    }
+
+    if timeout_seconds is not None:
+        spec['timeoutSeconds'] = timeout_seconds
+
+    return spec

agent/tools/hf/jobs/jobs_tool.py

@@ -0,0 +1,652 @@
+"""
+Hugging Face Jobs Tool - Manage compute jobs on Hugging Face
+
+Ported from: hf-mcp-server/packages/mcp/src/jobs/jobs-tool.ts
+"""
+import json
+from typing import Optional, Dict, Any, List, Literal
+from agent.tools.hf.types import ToolResult
+from agent.tools.hf.base import HfApiError
+from agent.tools.hf.jobs.api_client import JobsApiClient
+from agent.tools.hf.jobs.job_utils import create_job_spec
+from agent.tools.hf.jobs.uv_utils import resolve_uv_command, UV_DEFAULT_IMAGE
+from agent.tools.hf.utilities import (
+    format_jobs_table,
+    format_scheduled_jobs_table,
+    format_job_details,
+    format_scheduled_job_details,
+)
+
+
+# Hardware flavors
+CPU_FLAVORS = ['cpu-basic', 'cpu-upgrade', 'cpu-performance', 'cpu-xl']
+GPU_FLAVORS = [
+    'sprx8', 'zero-a10g', 't4-small', 't4-medium', 'l4x1', 'l4x4',
+    'l40sx1', 'l40sx4', 'l40sx8', 'a10g-small', 'a10g-large',
+    'a10g-largex2', 'a10g-largex4', 'a100-large', 'h100', 'h100x8'
+]
+SPECIALIZED_FLAVORS = ['inf2x6']
+ALL_FLAVORS = CPU_FLAVORS + GPU_FLAVORS + SPECIALIZED_FLAVORS
+
+# Operation names
+OperationType = Literal[
+    "run", "uv", "ps", "logs", "inspect", "cancel",
+    "scheduled run", "scheduled uv", "scheduled ps",
+    "scheduled inspect", "scheduled delete", "scheduled suspend", "scheduled resume"
+]
+
+# Constants
+DEFAULT_LOG_WAIT_SECONDS = 10
+
+
+class HfJobsTool:
+    """Tool for managing Hugging Face compute jobs"""
+
+    def __init__(self, hf_token: Optional[str] = None, namespace: Optional[str] = None):
+        self.hf_token = hf_token
+        self.client = JobsApiClient(hf_token, namespace)
+
+    async def execute(self, params: Dict[str, Any]) -> ToolResult:
+        """Execute the specified operation"""
+        operation = params.get('operation')
+        args = params.get('args', {})
+
+        # If no operation provided, return usage instructions
+        if not operation:
+            return self._show_help()
+
+        # Normalize operation name
+        operation = operation.lower()
+
+        # Check if help is requested
+        if args.get('help'):
+            return self._show_operation_help(operation)
+
+        try:
+            # Route to appropriate handler
+            if operation == "run":
+                return await self._run_job(args)
+            elif operation == "uv":
+                return await self._run_uv_job(args)
+            elif operation == "ps":
+                return await self._list_jobs(args)
+            elif operation == "logs":
+                return await self._get_logs(args)
+            elif operation == "inspect":
+                return await self._inspect_job(args)
+            elif operation == "cancel":
+                return await self._cancel_job(args)
+            elif operation == "scheduled run":
+                return await self._scheduled_run(args)
+            elif operation == "scheduled uv":
+                return await self._scheduled_uv(args)
+            elif operation == "scheduled ps":
+                return await self._list_scheduled_jobs(args)
+            elif operation == "scheduled inspect":
+                return await self._inspect_scheduled_job(args)
+            elif operation == "scheduled delete":
+                return await self._delete_scheduled_job(args)
+            elif operation == "scheduled suspend":
+                return await self._suspend_scheduled_job(args)
+            elif operation == "scheduled resume":
+                return await self._resume_scheduled_job(args)
+            else:
+                return {
+                    "formatted": f'Unknown operation: "{operation}"\n\n'
+                                'Available operations:\n'
+                                '- run, uv, ps, logs, inspect, cancel\n'
+                                '- scheduled run, scheduled uv, scheduled ps, scheduled inspect, '
+                                'scheduled delete, scheduled suspend, scheduled resume\n\n'
+                                'Call this tool with no operation for full usage instructions.',
+                    "totalResults": 0,
+                    "resultsShared": 0,
+                    "isError": True
+                }
+
+        except HfApiError as e:
+            error_message = f"API Error: {e.message}"
+            if e.response_body:
+                try:
+                    parsed = json.loads(e.response_body)
+                    formatted_body = json.dumps(parsed, indent=2)
+                    error_message += f"\n\nServer response:\n{formatted_body}"
+                except Exception:
+                    if len(e.response_body) < 500:
+                        error_message += f"\n\nServer response: {e.response_body}"
+
+            return {
+                "formatted": error_message,
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+        except Exception as e:
+            return {
+                "formatted": f"Error executing {operation}: {str(e)}",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+    def _show_help(self) -> ToolResult:
+        """Show usage instructions when tool is called with no arguments"""
+        cpu_flavors_list = ', '.join(CPU_FLAVORS)
+        gpu_flavors_list = ', '.join(GPU_FLAVORS)
+        specialized_flavors_list = ', '.join(SPECIALIZED_FLAVORS)
+
+        hardware_section = f"**CPU:** {cpu_flavors_list}\n"
+        if GPU_FLAVORS:
+            hardware_section += f"**GPU:** {gpu_flavors_list}\n"
+        if SPECIALIZED_FLAVORS:
+            hardware_section += f"**Specialized:** {specialized_flavors_list}"
+
+        usage_text = f"""# HuggingFace Jobs API
+
+Manage compute jobs on Hugging Face infrastructure.
+
+## Available Commands
+
+### Job Management
+- **run** - Run a job with a Docker image
+- **uv** - Run a Python script with UV (inline dependencies)
+- **ps** - List jobs
+- **logs** - Fetch job logs
+- **inspect** - Get detailed job information
+- **cancel** - Cancel a running job
+
+### Scheduled Jobs
+- **scheduled run** - Create a scheduled job
+- **scheduled uv** - Create a scheduled UV job
+- **scheduled ps** - List scheduled jobs
+- **scheduled inspect** - Get scheduled job details
+- **scheduled delete** - Delete a scheduled job
+- **scheduled suspend** - Pause a scheduled job
+- **scheduled resume** - Resume a suspended job
+
+## Examples
+
+### Run a simple job
+Call this tool with:
+```json
+{{
+  "operation": "run",
+  "args": {{
+    "image": "python:3.12",
+    "command": ["python", "-c", "print('Hello from HF Jobs!')"],
+    "flavor": "cpu-basic"
+  }}
+}}
+```
+
+### Run a Python script with UV
+Call this tool with:
+```json
+{{
+  "operation": "uv",
+  "args": {{
+    "script": "import random\\nprint(42 + random.randint(1, 5))"
+  }}
+}}
+```
+
+## Hardware Flavors
+
+{hardware_section}
+
+## Command Format Guidelines
+
+**Array format (default):**
+- Recommended for every command—JSON keeps arguments intact (URLs with `&`, spaces, etc.)
+- Use `["/bin/sh", "-lc", "..."]` when you need shell operators like `&&`, `|`, or redirections
+- Works with any language: Python, bash, node, npm, uv, etc.
+
+**String format (simple cases only):**
+- Still accepted for backwards compatibility, parsed with POSIX shell semantics
+- Rejects shell operators and can mis-handle characters such as `&`; switch to arrays when things turn complex
+- `$HF_TOKEN` stays literal—forward it via `secrets: {{ "HF_TOKEN": "$HF_TOKEN" }}`
+
+### Show command-specific help
+Call this tool with:
+```json
+{{"operation": "<operation>", "args": {{"help": true}}}}
+```
+
+## Tips
+
+- Jobs default to non-detached mode (tail logs for up to {DEFAULT_LOG_WAIT_SECONDS}s or until completion). Set `detach: true` to return immediately.
+- Prefer array commands to avoid shell parsing surprises
+- To access private Hub assets, include `secrets: {{ "HF_TOKEN": "$HF_TOKEN" }}` to inject your auth token.
+"""
+        return {
+            "formatted": usage_text,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    def _show_operation_help(self, operation: str) -> ToolResult:
+        """Show help for a specific operation"""
+        help_text = f"Help for operation: {operation}\n\nCall with appropriate arguments. Use the main help for examples."
+        return {
+            "formatted": help_text,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _run_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Create and run a job"""
+        # Create job spec from args
+        job_spec = create_job_spec({
+            'image': args.get('image', 'python:3.12'),
+            'command': args.get('command'),
+            'flavor': args.get('flavor', 'cpu-basic'),
+            'env': args.get('env'),
+            'secrets': args.get('secrets'),
+            'timeout': args.get('timeout', '30m'),
+            'hfToken': self.hf_token,
+        })
+
+        # Submit job
+        job = await self.client.run_job(job_spec, args.get('namespace'))
+
+        job_url = f"https://huggingface.co/jobs/{job['owner']['name']}/{job['id']}"
+
+        # If detached, return immediately
+        if args.get('detach', False):
+            response = f"""Job started successfully!
+
+**Job ID:** {job['id']}
+**Status:** {job['status']['stage']}
+**View at:** {job_url}
+
+To check logs, call this tool with `{{"operation": "logs", "args": {{"job_id": "{job['id']}"}}}}`
+To inspect, call this tool with `{{"operation": "inspect", "args": {{"job_id": "{job['id']}"}}}}`"""
+            return {
+                "formatted": response,
+                "totalResults": 1,
+                "resultsShared": 1
+            }
+
+        # Not detached - return job info and link to logs
+        response = f"""Job started: {job['id']}
+
+**Status:** {job['status']['stage']}
+**View logs at:** {job_url}
+
+Note: Logs are being collected. Check the job page for real-time logs.
+"""
+        return {
+            "formatted": response,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _run_uv_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Run job with UV package manager"""
+        # UV jobs use a standard UV image
+        image = UV_DEFAULT_IMAGE
+
+        # Build UV command
+        command = resolve_uv_command(args)
+
+        # Convert to run args
+        run_args = {
+            'image': image,
+            'command': command,
+            'flavor': args.get('flavor', 'cpu-basic'),
+            'env': args.get('env'),
+            'secrets': args.get('secrets'),
+            'timeout': args.get('timeout', '30m'),
+            'detach': args.get('detach', False),
+            'namespace': args.get('namespace'),
+        }
+
+        return await self._run_job(run_args)
+
+    async def _list_jobs(self, args: Dict[str, Any]) -> ToolResult:
+        """List user's jobs"""
+        # Fetch all jobs from API
+        all_jobs = await self.client.list_jobs(args.get('namespace'))
+
+        # Filter jobs
+        jobs = all_jobs
+
+        # Default: show only running jobs unless --all is specified
+        if not args.get('all', False):
+            jobs = [job for job in jobs if job['status']['stage'] == 'RUNNING']
+
+        # Apply status filter if specified
+        if args.get('status'):
+            status_filter = args['status'].upper()
+            jobs = [job for job in jobs if status_filter in job['status']['stage'].upper()]
+
+        # Format as markdown table
+        table = format_jobs_table(jobs)
+
+        if len(jobs) == 0:
+            if args.get('all', False):
+                return {
+                    "formatted": "No jobs found.",
+                    "totalResults": 0,
+                    "resultsShared": 0
+                }
+            return {
+                "formatted": 'No running jobs found. Use `{"args": {"all": true}}` to show all jobs.',
+                "totalResults": 0,
+                "resultsShared": 0
+            }
+
+        response = f"**Jobs ({len(jobs)} of {len(all_jobs)} total):**\n\n{table}"
+        return {
+            "formatted": response,
+            "totalResults": len(all_jobs),
+            "resultsShared": len(jobs)
+        }
+
+    async def _get_logs(self, args: Dict[str, Any]) -> ToolResult:
+        """Get logs for a job"""
+        job_id = args.get('job_id')
+        if not job_id:
+            return {
+                "formatted": "job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        # Get namespace for the logs URL
+        namespace = await self.client.get_namespace(args.get('namespace'))
+        job_url = f"https://huggingface.co/jobs/{namespace}/{job_id}"
+
+        # For now, direct users to the web interface for logs
+        # Full SSE streaming implementation would be more complex
+        response = f"""**Logs for job {job_id}**
+
+View real-time logs at: {job_url}
+
+Note: Full log streaming support is coming soon. Please use the web interface for now.
+"""
+        return {
+            "formatted": response,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _inspect_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Get detailed information about one or more jobs"""
+        job_id = args.get('job_id')
+        if not job_id:
+            return {
+                "formatted": "job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        job_ids = job_id if isinstance(job_id, list) else [job_id]
+
+        # Fetch all jobs
+        jobs = []
+        for jid in job_ids:
+            try:
+                job = await self.client.get_job(jid, args.get('namespace'))
+                jobs.append(job)
+            except Exception as e:
+                raise Exception(f"Failed to fetch job {jid}: {str(e)}")
+
+        formatted_details = format_job_details(jobs)
+        response = f"**Job Details** ({len(jobs)} job{'s' if len(jobs) > 1 else ''}):\n\n{formatted_details}"
+
+        return {
+            "formatted": response,
+            "totalResults": len(jobs),
+            "resultsShared": len(jobs)
+        }
+
+    async def _cancel_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Cancel a running job"""
+        job_id = args.get('job_id')
+        if not job_id:
+            return {
+                "formatted": "job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        await self.client.cancel_job(job_id, args.get('namespace'))
+
+        response = f"""✓ Job {job_id} has been cancelled.
+
+To verify, call this tool with `{{"operation": "inspect", "args": {{"job_id": "{job_id}"}}}}`"""
+
+        return {
+            "formatted": response,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _scheduled_run(self, args: Dict[str, Any]) -> ToolResult:
+        """Create a scheduled job"""
+        # Create job spec
+        job_spec = create_job_spec({
+            'image': args.get('image', 'python:3.12'),
+            'command': args.get('command'),
+            'flavor': args.get('flavor', 'cpu-basic'),
+            'env': args.get('env'),
+            'secrets': args.get('secrets'),
+            'timeout': args.get('timeout', '30m'),
+            'hfToken': self.hf_token,
+        })
+
+        # Create scheduled job spec
+        scheduled_spec = {
+            'schedule': args.get('schedule'),
+            'suspend': args.get('suspend', False),
+            'jobSpec': job_spec,
+        }
+
+        # Submit scheduled job
+        scheduled_job = await self.client.create_scheduled_job(scheduled_spec, args.get('namespace'))
+
+        response = f"""✓ Scheduled job created successfully!
+
+**Scheduled Job ID:** {scheduled_job['id']}
+**Schedule:** {scheduled_job['schedule']}
+**Suspended:** {'Yes' if scheduled_job.get('suspend') else 'No'}
+**Next Run:** {scheduled_job.get('nextRun', 'N/A')}
+
+To inspect, call this tool with `{{"operation": "scheduled inspect", "args": {{"scheduled_job_id": "{scheduled_job['id']}"}}}}`
+To list all, call this tool with `{{"operation": "scheduled ps"}}`"""
+
+        return {
+            "formatted": response,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _scheduled_uv(self, args: Dict[str, Any]) -> ToolResult:
+        """Create a scheduled UV job"""
+        # For UV, use standard UV image
+        image = UV_DEFAULT_IMAGE
+
+        # Build UV command
+        command = resolve_uv_command(args)
+
+        # Convert to scheduled run args
+        scheduled_run_args = {
+            'schedule': args.get('schedule'),
+            'suspend': args.get('suspend', False),
+            'image': image,
+            'command': command,
+            'flavor': args.get('flavor', 'cpu-basic'),
+            'env': args.get('env'),
+            'secrets': args.get('secrets'),
+            'timeout': args.get('timeout', '30m'),
+            'namespace': args.get('namespace'),
+        }
+
+        return await self._scheduled_run(scheduled_run_args)
+
+    async def _list_scheduled_jobs(self, args: Dict[str, Any]) -> ToolResult:
+        """List scheduled jobs"""
+        # Fetch all scheduled jobs
+        all_jobs = await self.client.list_scheduled_jobs(args.get('namespace'))
+
+        # Filter jobs
+        jobs = all_jobs
+
+        # Default: hide suspended jobs unless --all is specified
+        if not args.get('all', False):
+            jobs = [job for job in jobs if not job.get('suspend', False)]
+
+        # Format as markdown table
+        table = format_scheduled_jobs_table(jobs)
+
+        if len(jobs) == 0:
+            if args.get('all', False):
+                return {
+                    "formatted": "No scheduled jobs found.",
+                    "totalResults": 0,
+                    "resultsShared": 0
+                }
+            return {
+                "formatted": 'No active scheduled jobs found. Use `{"args": {"all": true}}` to show suspended jobs.',
+                "totalResults": 0,
+                "resultsShared": 0
+            }
+
+        response = f"**Scheduled Jobs ({len(jobs)} of {len(all_jobs)} total):**\n\n{table}"
+        return {
+            "formatted": response,
+            "totalResults": len(all_jobs),
+            "resultsShared": len(jobs)
+        }
+
+    async def _inspect_scheduled_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Get details of a scheduled job"""
+        scheduled_job_id = args.get('scheduled_job_id')
+        if not scheduled_job_id:
+            return {
+                "formatted": "scheduled_job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        job = await self.client.get_scheduled_job(scheduled_job_id, args.get('namespace'))
+        formatted_details = format_scheduled_job_details(job)
+
+        return {
+            "formatted": f"**Scheduled Job Details:**\n\n{formatted_details}",
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _delete_scheduled_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Delete a scheduled job"""
+        scheduled_job_id = args.get('scheduled_job_id')
+        if not scheduled_job_id:
+            return {
+                "formatted": "scheduled_job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        await self.client.delete_scheduled_job(scheduled_job_id, args.get('namespace'))
+
+        return {
+            "formatted": f"✓ Scheduled job {scheduled_job_id} has been deleted.",
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _suspend_scheduled_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Suspend a scheduled job"""
+        scheduled_job_id = args.get('scheduled_job_id')
+        if not scheduled_job_id:
+            return {
+                "formatted": "scheduled_job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        await self.client.suspend_scheduled_job(scheduled_job_id, args.get('namespace'))
+
+        response = f"""✓ Scheduled job {scheduled_job_id} has been suspended.
+
+To resume, call this tool with `{{"operation": "scheduled resume", "args": {{"scheduled_job_id": "{scheduled_job_id}"}}}}`"""
+
+        return {
+            "formatted": response,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+    async def _resume_scheduled_job(self, args: Dict[str, Any]) -> ToolResult:
+        """Resume a suspended scheduled job"""
+        scheduled_job_id = args.get('scheduled_job_id')
+        if not scheduled_job_id:
+            return {
+                "formatted": "scheduled_job_id is required",
+                "totalResults": 0,
+                "resultsShared": 0,
+                "isError": True
+            }
+
+        await self.client.resume_scheduled_job(scheduled_job_id, args.get('namespace'))
+
+        response = f"""✓ Scheduled job {scheduled_job_id} has been resumed.
+
+To inspect, call this tool with `{{"operation": "scheduled inspect", "args": {{"scheduled_job_id": "{scheduled_job_id}"}}}}`"""
+
+        return {
+            "formatted": response,
+            "totalResults": 1,
+            "resultsShared": 1
+        }
+
+
+# Tool specification for agent registration
+HF_JOBS_TOOL_SPEC = {
+    "name": "hf_jobs",
+    "description": (
+        "Manage Hugging Face CPU/GPU compute jobs. Run commands in Docker containers, "
+        "execute Python scripts with UV. List, schedule and monitor jobs/logs. "
+        "Call this tool with no operation for full usage instructions and examples."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "operation": {
+                "type": "string",
+                "enum": [
+                    "run", "uv", "ps", "logs", "inspect", "cancel",
+                    "scheduled run", "scheduled uv", "scheduled ps",
+                    "scheduled inspect", "scheduled delete", "scheduled suspend", "scheduled resume"
+                ],
+                "description": (
+                    "Operation to execute. Valid values: run, uv, ps, logs, inspect, cancel, "
+                    "scheduled run, scheduled uv, scheduled ps, scheduled inspect, scheduled delete, "
+                    "scheduled suspend, scheduled resume"
+                )
+            },
+            "args": {
+                "type": "object",
+                "description": "Operation-specific arguments as a JSON object",
+                "additionalProperties": True
+            }
+        }
+    }
+}
+
+
+async def hf_jobs_handler(arguments: Dict[str, Any]) -> tuple[str, bool]:
+    """Handler for agent tool router"""
+    try:
+        tool = HfJobsTool()
+        result = await tool.execute(arguments)
+        return result["formatted"], not result.get("isError", False)
+    except Exception as e:
+        return f"Error executing HF Jobs tool: {str(e)}", False

agent/tools/hf/jobs/uv_utils.py

@@ -0,0 +1,66 @@
+"""
+UV command utilities
+
+Ported from: hf-mcp-server/packages/mcp/src/jobs/commands/uv-utils.ts
+"""
+import base64
+from typing import List, Dict, Optional, Any
+
+
+UV_DEFAULT_IMAGE = 'ghcr.io/astral-sh/uv:python3.12-bookworm'
+
+
+def build_uv_command(script: str, args: Dict[str, Any]) -> List[str]:
+    """Build UV run command"""
+    parts = ['uv', 'run']
+
+    # Add dependencies
+    with_deps = args.get('with_deps', [])
+    if with_deps:
+        for dep in with_deps:
+            parts.extend(['--with', dep])
+
+    # Add python version
+    python = args.get('python')
+    if python:
+        parts.extend(['-p', python])
+
+    parts.append(script)
+
+    # Add script arguments
+    script_args = args.get('script_args', [])
+    if script_args:
+        parts.extend(script_args)
+
+    return parts
+
+
+def wrap_inline_script(script: str, args: Dict[str, Any]) -> str:
+    """Wrap inline script with base64 encoding for UV"""
+    encoded = base64.b64encode(script.encode('utf-8')).decode('utf-8')
+    base_command = build_uv_command('-', args)
+    # Shell quote the command parts
+    quoted_command = ' '.join(base_command)
+    return f'echo "{encoded}" | base64 -d | {quoted_command}'
+
+
+def resolve_uv_command(args: Dict[str, Any]) -> List[str]:
+    """Resolve UV command based on script source"""
+    script_source = args.get('script', '')
+
+    options = {
+        'with_deps': args.get('with_deps'),
+        'python': args.get('python'),
+        'script_args': args.get('script_args'),
+    }
+
+    # URL script
+    if script_source.startswith('http://') or script_source.startswith('https://'):
+        return build_uv_command(script_source, options)
+
+    # Inline multi-line script
+    if '\n' in script_source:
+        return ['/bin/sh', '-lc', wrap_inline_script(script_source, options)]
+
+    # File path or single-line script
+    return build_uv_command(script_source, options)

agent/tools/hf/types.py

@@ -0,0 +1,14 @@
+"""
+Types for Hugging Face tools
+
+Ported from: hf-mcp-server/packages/mcp/src/types/
+"""
+from typing import TypedDict, Optional
+
+
+class ToolResult(TypedDict, total=False):
+    """Result returned by HF tool operations"""
+    formatted: str
+    totalResults: int
+    resultsShared: int
+    isError: bool

agent/tools/hf/utilities.py

@@ -0,0 +1,133 @@
+"""
+Utility functions for Hugging Face tools
+
+Ported from: hf-mcp-server/packages/mcp/src/jobs/formatters.ts
+"""
+from typing import Any, Dict, List, Optional
+from datetime import datetime
+
+
+def truncate(text: str, max_length: int) -> str:
+    """Truncate a string to a maximum length with ellipsis"""
+    if len(text) <= max_length:
+        return text
+    return text[:max_length - 3] + "..."
+
+
+def format_date(date_str: Optional[str]) -> str:
+    """Format a date string to a readable format"""
+    if not date_str:
+        return "N/A"
+    try:
+        date = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
+        return date.strftime('%Y-%m-%d %H:%M:%S')
+    except Exception:
+        return date_str
+
+
+def format_command(command: Optional[List[str]]) -> str:
+    """Format command array as a single string"""
+    if not command or len(command) == 0:
+        return "N/A"
+    return " ".join(command)
+
+
+def get_image_or_space(job: Dict[str, Any]) -> str:
+    """Get image/space identifier from job"""
+    if job.get('spaceId'):
+        return job['spaceId']
+    if job.get('dockerImage'):
+        return job['dockerImage']
+    return "N/A"
+
+
+def format_jobs_table(jobs: List[Dict[str, Any]]) -> str:
+    """Format jobs as a markdown table"""
+    if len(jobs) == 0:
+        return "No jobs found."
+
+    # Calculate dynamic ID column width
+    longest_id_length = max(len(job['id']) for job in jobs)
+    id_column_width = max(longest_id_length, len('JOB ID'))
+
+    # Define column widths
+    col_widths = {
+        'id': id_column_width,
+        'image': 20,
+        'command': 30,
+        'created': 19,
+        'status': 12,
+    }
+
+    # Build header
+    header = f"| {'JOB ID'.ljust(col_widths['id'])} | {'IMAGE/SPACE'.ljust(col_widths['image'])} | {'COMMAND'.ljust(col_widths['command'])} | {'CREATED'.ljust(col_widths['created'])} | {'STATUS'.ljust(col_widths['status'])} |"
+    separator = f"|{'-' * (col_widths['id'] + 2)}|{'-' * (col_widths['image'] + 2)}|{'-' * (col_widths['command'] + 2)}|{'-' * (col_widths['created'] + 2)}|{'-' * (col_widths['status'] + 2)}|"
+
+    # Build rows
+    rows = []
+    for job in jobs:
+        job_id = job['id']
+        image = truncate(get_image_or_space(job), col_widths['image'])
+        command = truncate(format_command(job.get('command')), col_widths['command'])
+        created = truncate(format_date(job.get('createdAt')), col_widths['created'])
+        status = truncate(job['status']['stage'], col_widths['status'])
+
+        rows.append(f"| {job_id.ljust(col_widths['id'])} | {image.ljust(col_widths['image'])} | {command.ljust(col_widths['command'])} | {created.ljust(col_widths['created'])} | {status.ljust(col_widths['status'])} |")
+
+    return '\n'.join([header, separator] + rows)
+
+
+def format_scheduled_jobs_table(jobs: List[Dict[str, Any]]) -> str:
+    """Format scheduled jobs as a markdown table"""
+    if len(jobs) == 0:
+        return "No scheduled jobs found."
+
+    # Calculate dynamic ID column width
+    longest_id_length = max(len(job['id']) for job in jobs)
+    id_column_width = max(longest_id_length, len('ID'))
+
+    # Define column widths
+    col_widths = {
+        'id': id_column_width,
+        'schedule': 12,
+        'image': 18,
+        'command': 25,
+        'lastRun': 19,
+        'nextRun': 19,
+        'suspend': 9,
+    }
+
+    # Build header
+    header = f"| {'ID'.ljust(col_widths['id'])} | {'SCHEDULE'.ljust(col_widths['schedule'])} | {'IMAGE/SPACE'.ljust(col_widths['image'])} | {'COMMAND'.ljust(col_widths['command'])} | {'LAST RUN'.ljust(col_widths['lastRun'])} | {'NEXT RUN'.ljust(col_widths['nextRun'])} | {'SUSPENDED'.ljust(col_widths['suspend'])} |"
+    separator = f"|{'-' * (col_widths['id'] + 2)}|{'-' * (col_widths['schedule'] + 2)}|{'-' * (col_widths['image'] + 2)}|{'-' * (col_widths['command'] + 2)}|{'-' * (col_widths['lastRun'] + 2)}|{'-' * (col_widths['nextRun'] + 2)}|{'-' * (col_widths['suspend'] + 2)}|"
+
+    # Build rows
+    rows = []
+    for job in jobs:
+        job_id = job['id']
+        schedule = truncate(job['schedule'], col_widths['schedule'])
+        image = truncate(get_image_or_space(job['jobSpec']), col_widths['image'])
+        command = truncate(format_command(job['jobSpec'].get('command')), col_widths['command'])
+        last_run = truncate(format_date(job.get('lastRun')), col_widths['lastRun'])
+        next_run = truncate(format_date(job.get('nextRun')), col_widths['nextRun'])
+        suspend = 'Yes' if job.get('suspend') else 'No'
+
+        rows.append(f"| {job_id.ljust(col_widths['id'])} | {schedule.ljust(col_widths['schedule'])} | {image.ljust(col_widths['image'])} | {command.ljust(col_widths['command'])} | {last_run.ljust(col_widths['lastRun'])} | {next_run.ljust(col_widths['nextRun'])} | {suspend.ljust(col_widths['suspend'])} |")
+
+    return '\n'.join([header, separator] + rows)
+
+
+def format_job_details(jobs: Any) -> str:
+    """Format job details as JSON in a markdown code block"""
+    import json
+    job_array = jobs if isinstance(jobs, list) else [jobs]
+    json_str = json.dumps(job_array, indent=2)
+    return f"```json\n{json_str}\n```"
+
+
+def format_scheduled_job_details(jobs: Any) -> str:
+    """Format scheduled job details as JSON in a markdown code block"""
+    import json
+    job_array = jobs if isinstance(jobs, list) else [jobs]
+    json_str = json.dumps(job_array, indent=2)
+    return f"```json\n{json_str}\n```"

tests/tools/__init__.py

@@ -0,0 +1 @@
+"""Tests for agent tools"""

tests/tools/hf/__init__.py

@@ -0,0 +1 @@
+"""Tests for HF tools"""

tests/tools/hf/jobs/__init__.py

@@ -0,0 +1 @@
+"""Tests for HF Jobs tool"""

tests/tools/hf/jobs/test_jobs_tool.py

@@ -0,0 +1,252 @@
+"""
+Tests for HF Jobs Tool
+
+Tests the jobs tool implementation
+"""
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+from agent.tools.hf.jobs.jobs_tool import HfJobsTool, hf_jobs_handler
+
+
+@pytest.mark.asyncio
+async def test_show_help():
+    """Test that help message is shown when no operation specified"""
+    tool = HfJobsTool()
+    result = await tool.execute({})
+
+    assert "HuggingFace Jobs API" in result["formatted"]
+    assert "Available Commands" in result["formatted"]
+    assert result["totalResults"] == 1
+    assert not result.get("isError", False)
+
+
+@pytest.mark.asyncio
+async def test_show_operation_help():
+    """Test operation-specific help"""
+    tool = HfJobsTool()
+    result = await tool.execute({"operation": "run", "args": {"help": True}})
+
+    assert "Help for operation" in result["formatted"]
+    assert result["totalResults"] == 1
+
+
+@pytest.mark.asyncio
+async def test_invalid_operation():
+    """Test invalid operation handling"""
+    tool = HfJobsTool()
+    result = await tool.execute({"operation": "invalid_op"})
+
+    assert result.get("isError") == True
+    assert "Unknown operation" in result["formatted"]
+
+
+@pytest.mark.asyncio
+async def test_run_job_missing_command():
+    """Test run job with missing required parameter"""
+    tool = HfJobsTool()
+    result = await tool.execute({
+        "operation": "run",
+        "args": {"image": "python:3.12"}
+    })
+
+    assert result.get("isError") == True
+    assert "command parameter is required" in result["formatted"]
+
+
+@pytest.mark.asyncio
+async def test_list_jobs_mock():
+    """Test list jobs with mock API"""
+    tool = HfJobsTool()
+
+    # Mock the API client
+    with patch.object(tool.client, 'list_jobs', new_callable=AsyncMock) as mock_list:
+        mock_list.return_value = [
+            {
+                'id': 'test-job-1',
+                'status': {'stage': 'RUNNING'},
+                'command': ['echo', 'test'],
+                'createdAt': '2024-01-01T00:00:00Z',
+                'owner': {'name': 'test-user'}
+            },
+            {
+                'id': 'test-job-2',
+                'status': {'stage': 'COMPLETED'},
+                'command': ['python', 'script.py'],
+                'createdAt': '2024-01-01T01:00:00Z',
+                'owner': {'name': 'test-user'}
+            }
+        ]
+
+        # Test listing only running jobs (default)
+        result = await tool.execute({"operation": "ps"})
+
+        assert not result.get("isError", False)
+        assert "test-job-1" in result["formatted"]
+        assert "test-job-2" not in result["formatted"]  # COMPLETED jobs filtered out
+        assert result["totalResults"] == 2
+        assert result["resultsShared"] == 1
+
+        # Test listing all jobs
+        result = await tool.execute({"operation": "ps", "args": {"all": True}})
+
+        assert not result.get("isError", False)
+        assert "test-job-1" in result["formatted"]
+        assert "test-job-2" in result["formatted"]
+        assert result["totalResults"] == 2
+        assert result["resultsShared"] == 2
+
+
+@pytest.mark.asyncio
+async def test_inspect_job_mock():
+    """Test inspect job with mock API"""
+    tool = HfJobsTool()
+
+    with patch.object(tool.client, 'get_job', new_callable=AsyncMock) as mock_get:
+        mock_get.return_value = {
+            'id': 'test-job-1',
+            'status': {'stage': 'RUNNING'},
+            'command': ['echo', 'test'],
+            'createdAt': '2024-01-01T00:00:00Z',
+            'owner': {'name': 'test-user'},
+            'flavor': 'cpu-basic'
+        }
+
+        result = await tool.execute({
+            "operation": "inspect",
+            "args": {"job_id": "test-job-1"}
+        })
+
+        assert not result.get("isError", False)
+        assert "test-job-1" in result["formatted"]
+        assert "Job Details" in result["formatted"]
+        mock_get.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_cancel_job_mock():
+    """Test cancel job with mock API"""
+    tool = HfJobsTool()
+
+    with patch.object(tool.client, 'cancel_job', new_callable=AsyncMock) as mock_cancel:
+        mock_cancel.return_value = None
+
+        result = await tool.execute({
+            "operation": "cancel",
+            "args": {"job_id": "test-job-1"}
+        })
+
+        assert not result.get("isError", False)
+        assert "cancelled" in result["formatted"]
+        assert "test-job-1" in result["formatted"]
+        mock_cancel.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_handler():
+    """Test the handler function"""
+    with patch('agent.tools.hf.jobs.jobs_tool.HfJobsTool') as MockTool:
+        mock_tool_instance = MockTool.return_value
+        mock_tool_instance.execute = AsyncMock(return_value={
+            "formatted": "Test output",
+            "totalResults": 1,
+            "resultsShared": 1,
+            "isError": False
+        })
+
+        output, success = await hf_jobs_handler({"operation": "ps"})
+
+        assert success == True
+        assert "Test output" in output
+
+
+@pytest.mark.asyncio
+async def test_handler_error():
+    """Test handler with error"""
+    with patch('agent.tools.hf.jobs.jobs_tool.HfJobsTool') as MockTool:
+        MockTool.side_effect = Exception("Test error")
+
+        output, success = await hf_jobs_handler({})
+
+        assert success == False
+        assert "Error" in output
+
+
+@pytest.mark.asyncio
+async def test_scheduled_jobs_mock():
+    """Test scheduled jobs operations with mock API"""
+    tool = HfJobsTool()
+
+    # Test list scheduled jobs
+    with patch.object(tool.client, 'list_scheduled_jobs', new_callable=AsyncMock) as mock_list:
+        mock_list.return_value = [
+            {
+                'id': 'sched-job-1',
+                'schedule': '@daily',
+                'suspend': False,
+                'jobSpec': {
+                    'command': ['python', 'backup.py'],
+                    'dockerImage': 'python:3.12'
+                },
+                'nextRun': '2024-01-02T00:00:00Z'
+            }
+        ]
+
+        result = await tool.execute({"operation": "scheduled ps"})
+
+        assert not result.get("isError", False)
+        assert "sched-job-1" in result["formatted"]
+        assert "Scheduled Jobs" in result["formatted"]
+
+
+def test_job_utils():
+    """Test job utility functions"""
+    from agent.tools.hf.jobs.job_utils import parse_timeout, parse_image_source, parse_command
+
+    # Test timeout parsing
+    assert parse_timeout("5m") == 300
+    assert parse_timeout("2h") == 7200
+    assert parse_timeout("30s") == 30
+    assert parse_timeout("1d") == 86400
+
+    # Test image source parsing
+    result = parse_image_source("python:3.12")
+    assert result["dockerImage"] == "python:3.12"
+    assert result["spaceId"] is None
+
+    result = parse_image_source("https://huggingface.co/spaces/user/space")
+    assert result["dockerImage"] is None
+    assert result["spaceId"] == "user/space"
+
+    # Test command parsing
+    result = parse_command(["python", "script.py"])
+    assert result["command"] == ["python", "script.py"]
+
+    result = parse_command("python script.py")
+    assert result["command"] == ["python", "script.py"]
+
+
+def test_uv_utils():
+    """Test UV utility functions"""
+    from agent.tools.hf.jobs.uv_utils import build_uv_command, resolve_uv_command
+
+    # Test build UV command
+    command = build_uv_command("script.py", {})
+    assert command == ["uv", "run", "script.py"]
+
+    command = build_uv_command("script.py", {
+        "with_deps": ["requests", "numpy"],
+        "python": "3.12"
+    })
+    assert "uv" in command
+    assert "run" in command
+    assert "--with" in command
+    assert "requests" in command
+    assert "-p" in command
+    assert "3.12" in command
+
+    # Test resolve UV command
+    command = resolve_uv_command({"script": "https://example.com/script.py"})
+    assert "https://example.com/script.py" in command
+
+    command = resolve_uv_command({"script": "print('hello')"})
+    assert command == ["uv", "run", "print('hello')"]