Action Examples

This page provides complete, working examples of actions you can deploy on Universal API.

Hello World

The simplest possible action:

def handler(event, context):
    params = event.get("queryStringParameters", {}) or {}
    name = params.get("name", "World")

    return {
        "statusCode": 200,
        "body": {"message": f"Hello, {name}!"}
    }

Usage:

curl "https://api.universalapi.co/action/act-xxx?name=Developer"

URL Shortener Lookup

Look up where a shortened URL redirects:

import requests

def handler(event, context):
    params = event.get("queryStringParameters", {}) or {}
    url = params.get("url")

    if not url:
        return {
            "statusCode": 400,
            "body": {"error": "Missing 'url' parameter"}
        }

    try:
        # Follow redirects but capture the chain
        response = requests.head(url, allow_redirects=True, timeout=10)

        return {
            "statusCode": 200,
            "body": {
                "original_url": url,
                "final_url": response.url,
                "status_code": response.status_code,
                "redirect_count": len(response.history)
            }
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": {"error": str(e)}
        }

JSON Validator

Validate and format JSON:

import json

def handler(event, context):
    # Get JSON from body or query param
    body = event.get("body", "")
    if isinstance(body, str) and body:
        try:
            input_json = json.loads(body)
        except json.JSONDecodeError as e:
            return {
                "statusCode": 400,
                "body": {
                    "valid": False,
                    "error": f"Invalid JSON: {str(e)}",
                    "position": e.pos
                }
            }
    else:
        params = event.get("queryStringParameters", {}) or {}
        json_str = params.get("json", "")
        try:
            input_json = json.loads(json_str)
        except json.JSONDecodeError as e:
            return {
                "statusCode": 400,
                "body": {
                    "valid": False,
                    "error": f"Invalid JSON: {str(e)}"
                }
            }

    return {
        "statusCode": 200,
        "body": {
            "valid": True,
            "formatted": json.dumps(input_json, indent=2),
            "minified": json.dumps(input_json, separators=(",", ":"))
        }
    }

IP Geolocation

Look up geographic info for an IP address:

import requests

def handler(event, context):
    params = event.get("queryStringParameters", {}) or {}
    ip = params.get("ip")

    # Use requester's IP if none provided
    if not ip:
        ip = event.get("headers", {}).get("X-Forwarded-For", "").split(",")[0].strip()

    if not ip:
        return {
            "statusCode": 400,
            "body": {"error": "No IP address provided or detected"}
        }

    try:
        response = requests.get(f"http://ip-api.com/json/{ip}", timeout=5)
        data = response.json()

        return {
            "statusCode": 200,
            "body": {
                "ip": ip,
                "country": data.get("country"),
                "region": data.get("regionName"),
                "city": data.get("city"),
                "zip": data.get("zip"),
                "lat": data.get("lat"),
                "lon": data.get("lon"),
                "isp": data.get("isp"),
                "timezone": data.get("timezone")
            }
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": {"error": str(e)}
        }

Google Drive File Search

Search files in Google Drive (requires OAuth):

import requests

def handler(event, context):
    # Check OAuth
    if "keys" not in event or "google_oauth" not in event["keys"]:
        return {
            "statusCode": 401,
            "body": {"error": "Please connect your Google account"}
        }

    access_token = event["keys"]["google_oauth"]["keyValue"]
    params = event.get("queryStringParameters", {}) or {}

    # Search parameters
    query = params.get("query", "")
    mime_type = params.get("type")  # e.g., "application/pdf"
    max_results = int(params.get("limit", 10))

    # Build search query
    q_parts = []
    if query:
        q_parts.append(f"name contains '{query}'")
    if mime_type:
        q_parts.append(f"mimeType='{mime_type}'")
    q_parts.append("trashed=false")

    drive_query = " and ".join(q_parts)

    try:
        response = requests.get(
            "https://www.googleapis.com/drive/v3/files",
            headers={"Authorization": f"Bearer {access_token}"},
            params={
                "q": drive_query,
                "pageSize": max_results,
                "fields": "files(id,name,mimeType,size,modifiedTime,webViewLink)",
                "orderBy": "modifiedTime desc"
            },
            timeout=10
        )

        if response.status_code != 200:
            return {
                "statusCode": response.status_code,
                "body": {"error": response.text}
            }

        files = response.json().get("files", [])

        return {
            "statusCode": 200,
            "body": {
                "query": drive_query,
                "count": len(files),
                "files": files
            }
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": {"error": str(e)}
        }

Weather Lookup (with API Key)

Get weather data using a stored API key:

import requests

def handler(event, context):
    # Check for API key
    if "keys" not in event or "openweather" not in event["keys"]:
        return {
            "statusCode": 401,
            "body": {"error": "OpenWeather API key not configured"}
        }

    api_key = event["keys"]["openweather"]["keyValue"]
    params = event.get("queryStringParameters", {}) or {}
    city = params.get("city")

    if not city:
        return {
            "statusCode": 400,
            "body": {"error": "Missing 'city' parameter"}
        }

    try:
        response = requests.get(
            "https://api.openweathermap.org/data/2.5/weather",
            params={
                "q": city,
                "appid": api_key,
                "units": "metric"
            },
            timeout=10
        )

        if response.status_code != 200:
            return {
                "statusCode": response.status_code,
                "body": {"error": f"Weather API error: {response.text}"}
            }

        data = response.json()

        return {
            "statusCode": 200,
            "body": {
                "city": data.get("name"),
                "country": data.get("sys", {}).get("country"),
                "temperature": data.get("main", {}).get("temp"),
                "feels_like": data.get("main", {}).get("feels_like"),
                "humidity": data.get("main", {}).get("humidity"),
                "description": data.get("weather", [{}])[0].get("description"),
                "wind_speed": data.get("wind", {}).get("speed")
            }
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": {"error": str(e)}
        }

Text Statistics

Analyze text and return statistics:

import re
from collections import Counter

def handler(event, context):
    # Get text from body or query param
    body = event.get("body", "")
    if isinstance(body, dict):
        text = body.get("text", "")
    elif isinstance(body, str) and body:
        import json
        try:
            data = json.loads(body)
            text = data.get("text", "")
        except:
            text = body
    else:
        params = event.get("queryStringParameters", {}) or {}
        text = params.get("text", "")

    if not text:
        return {
            "statusCode": 400,
            "body": {"error": "No text provided"}
        }

    # Calculate statistics
    words = re.findall(r'\b\w+\b', text.lower())
    sentences = re.split(r'[.!?]+', text)
    sentences = [s.strip() for s in sentences if s.strip()]

    word_freq = Counter(words)

    return {
        "statusCode": 200,
        "body": {
            "characters": len(text),
            "characters_no_spaces": len(text.replace(" ", "")),
            "words": len(words),
            "sentences": len(sentences),
            "paragraphs": len([p for p in text.split("\n\n") if p.strip()]),
            "avg_word_length": round(sum(len(w) for w in words) / len(words), 2) if words else 0,
            "avg_sentence_length": round(len(words) / len(sentences), 2) if sentences else 0,
            "top_words": word_freq.most_common(10)
        }
    }

Lambda Context Inspector

Debug action that shows the full event structure:

import json

def handler(event, context):
    # Sanitize sensitive data
    safe_event = event.copy()

    # Mask key values
    if "keys" in safe_event:
        for key_name, key_data in safe_event["keys"].items():
            if "keyValue" in key_data:
                safe_event["keys"][key_name]["keyValue"] = "***MASKED***"
            if "secretKeyValue" in key_data:
                safe_event["keys"][key_name]["secretKeyValue"] = "***MASKED***"

    # Get context info
    context_info = {
        "function_name": getattr(context, "function_name", "unknown"),
        "memory_limit_mb": getattr(context, "memory_limit_in_mb", "unknown"),
        "aws_request_id": getattr(context, "aws_request_id", "unknown"),
        "remaining_time_ms": context.get_remaining_time_in_millis() if hasattr(context, "get_remaining_time_in_millis") else "unknown"
    }

    return {
        "statusCode": 200,
        "body": {
            "message": "Lambda execution context",
            "event": safe_event,
            "context": context_info
        }
    }

Using These Examples

1. Go to universalapi.co/actions
2. Click "Create Action"
3. Fill in name and description
4. Paste the code
5. Specify any required keys in "Keys Needed"
6. Click "Create"
7. Test your action!

Next Steps

• Creating Actions - Detailed creation guide
• OAuth Integration - Connect Google services
• Executing Actions - API reference