Create custom tool

[andsty]

Tasks

I want to create a tool to call specific api with a tac parameter to get results

Description
here is my tool

import requests
from typing import Optional, List, Dict, Union

def get_device_data(tac: Union[str, int]) -> Optional[List[Dict[str, str]]]:
    """
    Fetches device data from ksqlDB for a given TAC (Type Allocation Code).
    
    Args:
        tac (str or int): Type Allocation Code to query.
    
    Returns:
        Optional[List[Dict[str, str]]]: List of device data dictionaries or None if no results.
    """
    # Convert TAC to string and sanitize
    tac = str(tac).strip('"\' ')
    
    # Validate TAC
    if not tac.isdigit():
        print(f"Invalid TAC: '{tac}' must be a numeric string.")
        return None
    
    # ksqlDB API endpoint
    url = "http://zzzzzzz:8088/query"
    
    # Prepare request payload
    payload = {
        "ksql": f"SELECT * FROM GSMATABLE WHERE TAC = '{tac}';",
        "streamsProperties": {}
    }
    
    # Request headers
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    
    try:
        # Send POST request to ksqlDB
        response = requests.post(url, headers=headers, json=payload, timeout=10)
        response.raise_for_status()
        
        # Parse JSON response
        data = response.json()
        
        # Validate response structure
        if not data or not isinstance(data, list):
            print(f"No data found for TAC: {tac}")
            return None
        
        # Extract schema from response header
        if 'header' not in data[0]:
            print("Invalid response format: No header found")
            return None
        
        # Parse column names from schema
        schema = data[0]['header'].get('schema', '')
        if not schema:
            print("No schema information available")
            return None
        
        # Extract column labels
        schema_labels = []
        for col in schema.split(','):
            parts = col.strip().split(' ')
            if len(parts) >= 2:
                col_name = parts[0].strip('`')
                schema_labels.append(col_name)
        
        # Process result rows
        result_data = []
        for row_obj in data[1:]:
            columns = row_obj.get('row', {}).get('columns', [])
            if not columns:
                continue
            
            result = {
                label: str(value) 
                for label, value in zip(schema_labels, columns)
            }
            result_data.append(result)
        
        return result_data if result_data else None
        
    except requests.exceptions.RequestException as e:
        print(f"Network error when querying TAC {tac}: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON parsing error for TAC {tac}: {e}")
        return None
    except Exception as e:
        print(f"Unexpected error processing TAC {tac}: {e}")
        return None

If i run it from command line i get results. the actual query is select * from GSMATABLE where TAC='00100900';
However if i run it from bee ui i get null result for the same tac. just trying to understand the structure of the tool

[psschwei]

Not sure, but a couple of questions to start narrowing down the issue and help debug:

Do you know which block is returning the null result? Is it the # Validate response structure one?

What does url = "http://zzzzzzz:8088/query" point to (i.e. a service running on your machine, a remote API, another container, etc.)? Could possibly be an issue connecting with the service from within the stack...

Did you see anything in the logs?

If all else fails, it might be worth starting with a simple tool that just checks it can reach your DB endpoint and then add bits back in until you hit the error.

Hope that helps!