---
title: "Advanced Tool Implementation Patterns"
task: ""
lineage_type: import
upstream_source: https://github.com/mims-harvard/ToolUniverse/blob/e2520a96/skills/devtu-create-tool/references/advanced-patterns.md
upstream_sha: e2520a96
imported_at: 2026-06-26
prompt_class: prompt
upstream_changes: accepted
author: upstream
validated: false
---

# Advanced Tool Implementation Patterns

This reference covers advanced patterns and techniques for ToolUniverse tool development.

## Offline / Computational Tools

Some tools require no network access at all — they perform purely local calculations using
mathematical formulas, lookup tables, or parametric models.  These tools are faster, more
reliable, and easier to test than API-backed tools.

### Minimal Offline Tool Skeleton

```python
# my_calculator_tool.py
import math
from typing import Dict, Any, Optional
from tooluniverse.base_tool import BaseTool
from tooluniverse.tool_registry import register_tool

@register_tool("MyCalculatorTool")
class MyCalculatorTool(BaseTool):
    """
    Brief description.  Runs entirely offline — no network requests.
    """

    def __init__(self, tool_config: Dict[str, Any]):
        super().__init__(tool_config)
        fields = tool_config.get("fields", {})
        self.operation = fields.get("operation", "default_op")

    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            # operation can come from fields (set in JSON) or runtime arg
            op = arguments.get("operation") or self.operation

            def _get(key: str) -> Optional[float]:
                val = arguments.get(key)
                return float(val) if val is not None else None

            x = _get("x")
            if x is None:
                return {"status": "error", "message": "Missing required parameter: x"}

            result_value = x * 2          # your formula here

            return {
                "status": "success",
                "data": {
                    "operation": op,
                    "result": result_value,
                    "result_formatted": f"{result_value:.4g}",
                },
                "metadata": {
                    "note": "Runs offline — no network request.",
                    "formula": "result = x × 2",
                },
            }
        except ValueError as e:
            return {"status": "error", "message": str(e)}
        except Exception as e:
            return {"status": "error", "message": f"Unexpected error: {str(e)}"}
```

Key rules for offline tools:
- Return `{"status": "error", "message": "..."}` — use `"message"` (not `"error"`) to match the circuit plugin convention.
- Include a `"metadata"` block that names the formula(s) used and notes "Runs offline".
- Always `float(val)` when extracting numeric arguments from the dict (they may arrive as strings).
- Use `ValueError` for domain validation; let the outer `except Exception` catch everything else.

### SI Formatting Helper

Include a local `_fmt_si()` helper in every offline calculation tool for consistent output:

```python
def _fmt_si(value: float, unit: str) -> str:
    """Format a numeric value with SI prefix and unit label."""
    abs_v = abs(value)
    if abs_v == 0:
        return f"0 {unit}"
    if abs_v >= 1e9:  return f"{value / 1e9:.4g} G{unit}"
    if abs_v >= 1e6:  return f"{value / 1e6:.4g} M{unit}"
    if abs_v >= 1e3:  return f"{value / 1e3:.4g} k{unit}"
    if abs_v >= 1:    return f"{value:.4g} {unit}"
    if abs_v >= 1e-3: return f"{value * 1e3:.4g} m{unit}"
    if abs_v >= 1e-6: return f"{value * 1e6:.4g} µ{unit}"
    if abs_v >= 1e-9: return f"{value * 1e9:.4g} n{unit}"
    return f"{value * 1e12:.4g} p{unit}"
```

Always include both the raw numeric field and a `_formatted` string sibling so LLMs can
display human-readable values without reformatting:

```python
"data": {
    "delay_ps": 34.0,
    "delay_formatted": "34 ps",        # human-readable sibling
    "frequency_Hz": 4.66e9,
    "frequency_formatted": "4.66 GHz", # human-readable sibling
}
```

### Technology-Node Lookup Tables

Many chip-design quantities are tabulated per process node.  Use a `Dict[int, ...]` keyed
by node in nm with a `_nearest_node()` helper that falls back to the closest entry:

```python
# Empirical FO4 delay values in ps, one per node
_FO4_TABLE: Dict[int, float] = {
    180: 250.0,
    130: 170.0,
    90:  115.0,
    65:   80.0,
    45:   55.0,
    32:   40.0,
    28:   34.0,
    20:   26.0,
    16:   20.0,
    10:   15.0,
    7:    11.0,
}

_SORTED_NODES = sorted(_FO4_TABLE.keys())

def _nearest_node(node_nm: float) -> int:
    """Return the nearest supported technology node for a given nm value."""
    return min(_SORTED_NODES, key=lambda n: abs(n - node_nm))
```

When the user's requested node is not in the table, include a `"warning"` field (not an
error) in the data dict and continue with the nearest match:

```python
nearest = _nearest_node(node_nm)
fo4_ps  = _FO4_TABLE[nearest]

warning = None
if nearest != int(node_nm):
    warning = (
        f"Node {node_nm} nm not in table; "
        f"using nearest supported node {nearest} nm."
    )

result = {
    "node_nm": nearest,
    "fo4_delay_ps": fo4_ps,
    # ...
}
if warning:
    result["warning"] = warning      # add only when needed
```

Also expose `"supported_nodes_nm": _SORTED_NODES` in the response so callers know which
values are available without reading the source code.

### Warning Fields vs Error Returns

Use these two patterns consistently:

| Situation | Pattern |
|---|---|
| Input is technically valid but uses a fallback (e.g. nearest node) | `data["warning"] = "..."` — return `status: success` |
| Input violates a hard physical constraint (e.g. negative capacitance) | `raise ValueError("...")` — caught → `status: error, message: ...` |
| A computed threshold is exceeded (e.g. ground bounce > 100 mV) | `data["warning"] = "..."` + `data["exceeds_threshold"] = True` — still `status: success` |

Example — threshold warning pattern:

```python
THRESHOLD_V = 0.100  # 100 mV

v_noise = L * n * di_dt   # computation
exceeds = v_noise > THRESHOLD_V

result = {
    "ground_bounce_V":    v_noise,
    "ground_bounce_mV":   v_noise * 1e3,
    "exceeds_threshold":  exceeds,
}
if exceeds:
    result["warning"] = (
        f"Ground bounce {v_noise * 1e3:.1f} mV exceeds "
        f"{THRESHOLD_V * 1e3:.0f} mV threshold."
    )
```

### Multi-Operation Offline Tool (fields.operation dispatch)

For tools that share the same physics domain but have multiple distinct computations,
use the `fields.operation` dispatch pattern.  The JSON config stores the default operation
in `"fields": { "operation": "op_name" }`, and the runtime argument can override it:

```python
# In JSON config (one entry per operation):
{
  "name": "Circuit_fo4_delay",
  "type": "FO4DelayTool",
  "fields": { "operation": "lookup" },    # default for this tool entry
  ...
}

# In __init__:
def __init__(self, tool_config):
    super().__init__(tool_config)
    self.operation = tool_config.get("fields", {}).get("operation", "lookup")

# In run():
op = arguments.get("operation") or self.operation   # runtime arg takes priority
if op == "lookup":
    ...
elif op == "estimate_path":
    ...
else:
    return {"status": "error", "message": f"Unknown operation '{op}'.  Valid: ..."}
```

This lets you expose a single Python class and multiple JSON entries (one per default
operation) — or a single JSON entry that accepts the operation as a runtime argument.

## Caching Strategies

### Simple LRU Cache

```python
from functools import lru_cache
import json

@register_tool("CachedAPITool")
class CachedAPITool(BaseTool):
    """Tool with response caching."""
    
    @lru_cache(maxsize=128)
    def _cached_request(self, url: str, params_json: str):
        """Cache API responses using LRU cache."""
        params = json.loads(params_json)
        response = requests.get(url, params=params)
        return response.json()
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        url = "https://api.example.com/data"
        
        # Convert params to JSON string for cache key
        params_json = json.dumps(arguments, sort_keys=True)
        
        try:
            data = self._cached_request(url, params_json)
            return {"status": "success", "data": data}
        except Exception as e:
            return {"status": "error", "error": str(e)}
```

### Time-Based Cache

```python
from datetime import datetime, timedelta
from typing import Optional

@register_tool("TimeCachedTool")
class TimeCachedTool(BaseTool):
    """Tool with time-based caching."""
    
    def __init__(self):
        super().__init__()
        self._cache = {}
        self._cache_timeout = timedelta(minutes=15)
    
    def _get_cached(self, key: str) -> Optional[Dict]:
        """Get cached value if not expired."""
        if key in self._cache:
            data, timestamp = self._cache[key]
            if datetime.now() - timestamp < self._cache_timeout:
                return data
            else:
                del self._cache[key]
        return None
    
    def _set_cached(self, key: str, data: Dict):
        """Store value in cache with timestamp."""
        self._cache[key] = (data, datetime.now())
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        cache_key = json.dumps(arguments, sort_keys=True)
        
        # Check cache first
        cached = self._get_cached(cache_key)
        if cached:
            return {
                "status": "success",
                "data": cached,
                "cached": True
            }
        
        # Fetch fresh data
        try:
            response = requests.get(url, params=arguments)
            data = response.json()
            
            # Store in cache
            self._set_cached(cache_key, data)
            
            return {
                "status": "success",
                "data": data,
                "cached": False
            }
        except Exception as e:
            return {"status": "error", "error": str(e)}
```

## Batch Processing

### Batch Request Tool

```python
from typing import List

@register_tool("BatchTool")
class BatchTool(BaseTool):
    """Process multiple requests in a single call."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        ids = arguments.get('ids', [])
        batch_size = arguments.get('batch_size', 10)
        
        if not ids:
            return {
                "status": "error",
                "error": "ids parameter is required"
            }
        
        results = []
        errors = []
        
        # Process in batches
        for i in range(0, len(ids), batch_size):
            batch = ids[i:i + batch_size]
            
            try:
                batch_results = self._fetch_batch(batch)
                results.extend(batch_results)
            except Exception as e:
                errors.append({
                    "batch": batch,
                    "error": str(e)
                })
        
        return {
            "status": "success" if not errors else "partial",
            "count": len(results),
            "results": results,
            "errors": errors if errors else None
        }
    
    def _fetch_batch(self, ids: List[str]) -> List[Dict]:
        """Fetch data for a batch of IDs."""
        response = requests.post(
            "https://api.example.com/batch",
            json={"ids": ids},
            timeout=60
        )
        response.raise_for_status()
        return response.json()
```

### Parallel Requests

```python
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

@register_tool("ParallelTool")
class ParallelTool(BaseTool):
    """Execute multiple requests in parallel."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        ids = arguments.get('ids', [])
        max_workers = min(arguments.get('max_workers', 5), 10)
        
        results = []
        errors = []
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all requests
            future_to_id = {
                executor.submit(self._fetch_single, id_): id_
                for id_ in ids
            }
            
            # Collect results as they complete
            for future in as_completed(future_to_id):
                id_ = future_to_id[future]
                try:
                    result = future.result()
                    results.append(result)
                except Exception as e:
                    errors.append({
                        "id": id_,
                        "error": str(e)
                    })
        
        return {
            "status": "success" if not errors else "partial",
            "count": len(results),
            "results": results,
            "errors": errors if errors else None
        }
    
    def _fetch_single(self, id_: str) -> Dict:
        """Fetch data for a single ID."""
        response = requests.get(
            f"https://api.example.com/items/{id_}",
            timeout=30
        )
        response.raise_for_status()
        return response.json()
```

## Streaming and Pagination

### Auto-Pagination Tool

```python
@register_tool("AutoPaginationTool")
class AutoPaginationTool(BaseTool):
    """Automatically fetch all pages."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        max_pages = arguments.get('max_pages', 10)
        
        all_results = []
        page = 1
        
        while page <= max_pages:
            try:
                response = requests.get(
                    "https://api.example.com/search",
                    params={
                        'query': query,
                        'page': page,
                        'page_size': 100
                    },
                    timeout=30
                )
                response.raise_for_status()
                data = response.json()
                
                results = data.get('results', [])
                all_results.extend(results)
                
                # Stop if no more results
                if not data.get('next'):
                    break
                
                page += 1
                
            except Exception as e:
                return {
                    "status": "error",
                    "error": f"Failed on page {page}: {str(e)}",
                    "partial_results": all_results
                }
        
        return {
            "status": "success",
            "count": len(all_results),
            "pages": page,
            "results": all_results
        }
```

### Cursor-Based Pagination

```python
@register_tool("CursorPaginationTool")
class CursorPaginationTool(BaseTool):
    """Handle cursor-based pagination."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        cursor = arguments.get('cursor')
        page_size = arguments.get('page_size', 20)
        
        params = {'page_size': page_size}
        if cursor:
            params['cursor'] = cursor
        
        try:
            response = requests.get(
                "https://api.example.com/items",
                params=params,
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            return {
                "status": "success",
                "count": len(data.get('results', [])),
                "next_cursor": data.get('next_cursor'),
                "has_more": data.get('has_more', False),
                "results": data.get('results', [])
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

## Authentication Patterns

### API Key Authentication

```python
import os

@register_tool("APIKeyTool")
class APIKeyTool(BaseTool):
    """Tool with API key authentication."""
    
    def __init__(self):
        super().__init__()
        self.api_key = os.environ.get('API_KEY')
        if not self.api_key:
            raise ValueError("API_KEY environment variable not set")
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            response = requests.get(
                "https://api.example.com/data",
                headers={"X-API-Key": self.api_key},
                params=arguments,
                timeout=30
            )
            response.raise_for_status()
            
            return {
                "status": "success",
                "data": response.json()
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

### OAuth Token Authentication

```python
@register_tool("OAuthTool")
class OAuthTool(BaseTool):
    """Tool with OAuth token authentication."""
    
    def __init__(self):
        super().__init__()
        self.token = self._get_token()
    
    def _get_token(self) -> str:
        """Get or refresh OAuth token."""
        client_id = os.environ.get('CLIENT_ID')
        client_secret = os.environ.get('CLIENT_SECRET')
        
        response = requests.post(
            "https://api.example.com/oauth/token",
            data={
                'grant_type': 'client_credentials',
                'client_id': client_id,
                'client_secret': client_secret
            }
        )
        response.raise_for_status()
        return response.json()['access_token']
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            response = requests.get(
                "https://api.example.com/data",
                headers={"Authorization": f"Bearer {self.token}"},
                params=arguments,
                timeout=30
            )
            response.raise_for_status()
            
            return {
                "status": "success",
                "data": response.json()
            }
        except requests.HTTPError as e:
            if e.response.status_code == 401:
                # Token expired, refresh and retry
                self.token = self._get_token()
                return self.run(arguments)
            return {
                "status": "error",
                "error": str(e)
            }
```

## Data Transformation

### Field Selection and Projection

```python
@register_tool("ProjectionTool")
class ProjectionTool(BaseTool):
    """Tool with field selection support."""
    
    ALLOWED_FIELDS = {
        'id', 'name', 'description', 'created_date',
        'updated_date', 'status', 'author', 'tags'
    }
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        id_ = arguments.get('id')
        fields = arguments.get('fields', ['id', 'name'])
        
        # Validate fields
        invalid = set(fields) - self.ALLOWED_FIELDS
        if invalid:
            return {
                "status": "error",
                "error": f"Invalid fields: {invalid}",
                "allowed_fields": list(self.ALLOWED_FIELDS)
            }
        
        try:
            # Fetch full data
            response = requests.get(
                f"https://api.example.com/items/{id_}",
                timeout=30
            )
            response.raise_for_status()
            full_data = response.json()
            
            # Project only requested fields
            projected = {
                k: v for k, v in full_data.items()
                if k in fields
            }
            
            return {
                "status": "success",
                "data": projected
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

### Data Filtering

```python
@register_tool("FilterTool")
class FilterTool(BaseTool):
    """Tool with client-side filtering."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        filters = arguments.get('filters', {})
        
        try:
            # Fetch data
            response = requests.get(
                "https://api.example.com/search",
                params={'query': query},
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            # Apply filters
            results = data.get('results', [])
            filtered = self._apply_filters(results, filters)
            
            return {
                "status": "success",
                "count": len(filtered),
                "total": len(results),
                "results": filtered
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
    
    def _apply_filters(self, results: List[Dict], filters: Dict) -> List[Dict]:
        """Apply filters to results."""
        filtered = results
        
        # Status filter
        if 'status' in filters:
            status = filters['status']
            filtered = [r for r in filtered if r.get('status') == status]
        
        # Date range filter
        if 'date_from' in filters:
            date_from = filters['date_from']
            filtered = [
                r for r in filtered
                if r.get('date', '') >= date_from
            ]
        
        if 'date_to' in filters:
            date_to = filters['date_to']
            filtered = [
                r for r in filtered
                if r.get('date', '') <= date_to
            ]
        
        # Tag filter
        if 'tags' in filters:
            required_tags = set(filters['tags'])
            filtered = [
                r for r in filtered
                if required_tags.issubset(set(r.get('tags', [])))
            ]
        
        return filtered
```

## GraphQL Integration

### GraphQL Query Tool

```python
@register_tool("GraphQLTool")
class GraphQLTool(BaseTool):
    """Execute GraphQL queries."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        variables = arguments.get('variables', {})
        
        if not query:
            return {
                "status": "error",
                "error": "query parameter is required"
            }
        
        try:
            response = requests.post(
                "https://api.example.com/graphql",
                json={
                    'query': query,
                    'variables': variables
                },
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            # Check for GraphQL errors
            if 'errors' in result:
                return {
                    "status": "error",
                    "error": "GraphQL query failed",
                    "details": result['errors']
                }
            
            return {
                "status": "success",
                "data": result.get('data')
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

### GraphQL with Predefined Queries

```python
@register_tool("PredefinedGraphQLTool")
class PredefinedGraphQLTool(BaseTool):
    """GraphQL tool with predefined queries."""
    
    QUERIES = {
        'get_drug': """
            query GetDrug($id: ID!) {
                drug(id: $id) {
                    id
                    name
                    description
                    manufacturer
                }
            }
        """,
        'search_drugs': """
            query SearchDrugs($query: String!, $limit: Int) {
                searchDrugs(query: $query, limit: $limit) {
                    id
                    name
                    manufacturer
                }
            }
        """
    }
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        operation = arguments.get('operation')
        variables = arguments.get('variables', {})
        
        if operation not in self.QUERIES:
            return {
                "status": "error",
                "error": f"Unknown operation: {operation}",
                "available_operations": list(self.QUERIES.keys())
            }
        
        query = self.QUERIES[operation]
        
        try:
            response = requests.post(
                "https://api.example.com/graphql",
                json={
                    'query': query,
                    'variables': variables
                },
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            if 'errors' in result:
                return {
                    "status": "error",
                    "error": "GraphQL query failed",
                    "details": result['errors']
                }
            
            return {
                "status": "success",
                "data": result.get('data')
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

## Rate Limiting

### Simple Rate Limiter

```python
from time import time, sleep
from collections import deque

@register_tool("RateLimitedTool")
class RateLimitedTool(BaseTool):
    """Tool with rate limiting."""
    
    def __init__(self):
        super().__init__()
        self.requests = deque()
        self.max_requests = 10  # 10 requests
        self.time_window = 60   # per 60 seconds
    
    def _wait_if_needed(self):
        """Wait if rate limit would be exceeded."""
        now = time()
        
        # Remove requests outside time window
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()
        
        # Wait if at limit
        if len(self.requests) >= self.max_requests:
            sleep_time = self.requests[0] + self.time_window - now
            if sleep_time > 0:
                sleep(sleep_time)
            self.requests.popleft()
        
        # Record this request
        self.requests.append(now)
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        self._wait_if_needed()
        
        try:
            response = requests.get(
                "https://api.example.com/data",
                params=arguments,
                timeout=30
            )
            response.raise_for_status()
            
            return {
                "status": "success",
                "data": response.json()
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

## Webhook and Async Operations

### Async Job Polling Tool

```python
@register_tool("AsyncJobTool")
class AsyncJobTool(BaseTool):
    """Tool that polls for async job completion."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        timeout = arguments.get('timeout', 300)  # 5 minutes default
        poll_interval = arguments.get('poll_interval', 5)  # 5 seconds
        
        try:
            # Start async job
            response = requests.post(
                "https://api.example.com/jobs",
                json={'query': query},
                timeout=30
            )
            response.raise_for_status()
            job_data = response.json()
            job_id = job_data['job_id']
            
            # Poll for completion
            start_time = time()
            while time() - start_time < timeout:
                status_response = requests.get(
                    f"https://api.example.com/jobs/{job_id}",
                    timeout=30
                )
                status_response.raise_for_status()
                status = status_response.json()
                
                if status['state'] == 'completed':
                    return {
                        "status": "success",
                        "job_id": job_id,
                        "data": status['result']
                    }
                elif status['state'] == 'failed':
                    return {
                        "status": "error",
                        "error": "Job failed",
                        "detail": status.get('error')
                    }
                
                sleep(poll_interval)
            
            return {
                "status": "timeout",
                "error": f"Job did not complete within {timeout} seconds",
                "job_id": job_id
            }
            
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
```

## Multi-Source Aggregation

### Aggregation Tool

```python
@register_tool("AggregationTool")
class AggregationTool(BaseTool):
    """Aggregate data from multiple sources."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        sources = arguments.get('sources', ['source1', 'source2', 'source3'])
        
        results = []
        errors = []
        
        for source in sources:
            try:
                data = self._fetch_from_source(source, query)
                results.extend(data)
            except Exception as e:
                errors.append({
                    "source": source,
                    "error": str(e)
                })
        
        # Deduplicate by ID
        seen_ids = set()
        unique_results = []
        for result in results:
            id_ = result.get('id')
            if id_ not in seen_ids:
                seen_ids.add(id_)
                unique_results.append(result)
        
        return {
            "status": "success" if not errors else "partial",
            "count": len(unique_results),
            "sources_queried": len(sources),
            "sources_failed": len(errors),
            "results": unique_results,
            "errors": errors if errors else None
        }
    
    def _fetch_from_source(self, source: str, query: str) -> List[Dict]:
        """Fetch data from a specific source."""
        url_map = {
            'source1': 'https://api1.example.com/search',
            'source2': 'https://api2.example.com/query',
            'source3': 'https://api3.example.com/find'
        }
        
        url = url_map.get(source)
        if not url:
            raise ValueError(f"Unknown source: {source}")
        
        response = requests.get(url, params={'q': query}, timeout=30)
        response.raise_for_status()
        
        return response.json().get('results', [])
```