Functions

Handler decorators and function execution in the AGNT5 Python SDK

Functions are the core building blocks of AGNT5 applications. Use the @function decorator to register Python callables as invokable components that can be discovered and executed by the platform.

Basic Usage

Simple Function

from agnt5 import function

@function()
def greet_user(name: str) -> str:
    """Greet a user by name."""
    return f"Hello, {name}!"

Named Function

Override the registered name:

@function("math.add")
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

Function with Context

Access execution metadata through the context parameter:

from agnt5 import function
from agnt5.components import ExecutionContext

@function()
def context_aware(ctx: ExecutionContext, data: dict) -> dict:
    """Process data with execution context."""
    return {
        "invocation_id": ctx.invocation_id,
        "service_name": ctx.metadata.get("service_name"),
        "processed_data": data,
        "component_type": ctx.component_type.value
    }

Decorator Parameters

`function(name=None)`

Parameter	Type	Description
`name`	`str \| None`	Override the registered function name. Defaults to the original function name.

# Uses function name "process_data"
@function()
def process_data(data: dict) -> dict:
    return data

# Uses custom name "data_processor"
@function("data_processor")
def process_data(data: dict) -> dict:
    return data

Handler Signatures

AGNT5 supports flexible function signatures to accommodate different use cases.

Without Context

For simple stateless functions:

@function()
def calculate_tax(amount: float, rate: float) -> float:
    return amount * rate

@function()
def format_message(template: str, **kwargs) -> str:
    return template.format(**kwargs)

With Context

When you need access to invocation metadata:

@function()
def audit_handler(ctx: ExecutionContext, action: str, data: dict) -> dict:
    """Handler that logs audit information."""
    import logging

    logger = logging.getLogger(__name__)
    logger.info(f"Audit: {action} from {ctx.invocation_id}")

    return {
        "action": action,
        "invocation_id": ctx.invocation_id,
        "data": data,
        "timestamp": time.time()
    }

Async Functions

AGNT5 supports both synchronous and asynchronous functions:

Async Handler

import asyncio
from agnt5 import function

@function()
async def async_processor(data: dict) -> dict:
    """Async processing with I/O operations."""
    # Simulate async I/O
    await asyncio.sleep(0.1)

    # Async HTTP request example
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.example.com/process",
            json=data
        )
        external_result = response.json()

    return {
        "original": data,
        "external": external_result,
        "processed_at": datetime.utcnow().isoformat()
    }

Async with Context

@function()
async def async_context_handler(ctx: ExecutionContext, query: str) -> dict:
    """Async handler with context access."""

    # Use context for correlation
    correlation_id = ctx.metadata.get("correlation_id", ctx.invocation_id)

    # Async database query
    result = await database.execute(
        "SELECT * FROM items WHERE name LIKE %s",
        f"%{query}%"
    )

    return {
        "correlation_id": correlation_id,
        "query": query,
        "results": [dict(row) for row in result]
    }

Streaming Functions

For functions that need to return multiple responses over time:

Basic Streaming

from agnt5 import function

@function(streaming=True)
async def stream_data(count: int):
    """Stream multiple data chunks."""
    for i in range(count):
        yield {
            "chunk": i,
            "data": f"Data chunk {i}",
            "timestamp": time.time()
        }
        await asyncio.sleep(0.1)  # Simulate processing time

Streaming with Context

@function(streaming=True)
async def stream_with_context(ctx: ExecutionContext, query: str):
    """Stream search results progressively."""
    search_id = ctx.invocation_id

    # Stream results as they're found
    async for result in search_engine.stream_search(query):
        yield {
            "search_id": search_id,
            "result": result,
            "timestamp": time.time()
        }

Function Metadata

The SDK automatically captures and provides metadata about registered functions:

Inspecting Functions

from agnt5.decorators import get_registered_functions, get_function_metadata

# Get all registered functions
functions = get_registered_functions()
print(f"Registered functions: {list(functions.keys())}")

# Get metadata for a specific function
@function()
def sample_function(name: str, age: int = 25) -> dict:
    return {"name": name, "age": age}

metadata = get_function_metadata(sample_function)
print(metadata)

Output:

{
    "name": "sample_function",
    "type": "function",
    "parameters": [
        {"name": "name", "type": "str", "required": True},
        {"name": "age", "type": "int", "required": False, "default": 25}
    ],
    "return_type": "dict"
}

Runtime Annotations

The decorator adds runtime annotations to functions:

@function()
def annotated_function(data: str) -> str:
    return data.upper()

# Check annotations
print(annotated_function._agnt5_handler_name)  # "annotated_function"
print(annotated_function._agnt5_is_function)   # True

Error Handling

Basic Error Handling

@function()
def safe_divider(a: float, b: float) -> dict:
    """Safely divide two numbers."""
    try:
        if b == 0:
            return {"error": "Division by zero", "result": None}

        result = a / b
        return {"result": result, "error": None}

    except Exception as e:
        return {"error": str(e), "result": None}

Context-Aware Error Handling

import logging
from agnt5 import function
from agnt5.components import ExecutionContext

@function()
def robust_handler(ctx: ExecutionContext, data: dict) -> dict:
    """Handler with comprehensive error handling."""
    logger = logging.getLogger(__name__)

    try:
        # Log the invocation
        logger.info(f"Processing invocation {ctx.invocation_id}")

        # Validate input
        if not isinstance(data, dict):
            raise ValueError("Input must be a dictionary")

        required_fields = ["id", "name"]
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            raise ValueError(f"Missing required fields: {missing_fields}")

        # Process data
        result = {
            "processed": True,
            "id": data["id"],
            "name": data["name"].upper(),
            "invocation_id": ctx.invocation_id
        }

        logger.info(f"Successfully processed {data['id']}")
        return result

    except ValueError as e:
        logger.warning(f"Validation error in {ctx.invocation_id}: {e}")
        return {"error": f"Validation error: {e}", "result": None}

    except Exception as e:
        logger.error(f"Unexpected error in {ctx.invocation_id}: {e}")
        return {"error": "Internal error", "result": None}

Type Annotations

Use Python type hints for better documentation and validation:

Basic Types

from typing import Dict, List, Optional, Union

@function()
def typed_handler(
    name: str,
    age: int,
    tags: List[str],
    metadata: Optional[Dict[str, any]] = None
) -> Dict[str, Union[str, int, List[str]]]:
    """Handler with comprehensive type annotations."""
    return {
        "name": name,
        "age": age,
        "tags": tags,
        "has_metadata": metadata is not None
    }

Pydantic Models

For complex data validation:

from pydantic import BaseModel, Field
from typing import Optional
from agnt5 import function

class UserRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: Optional[int] = Field(None, ge=0, le=150)

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    age: Optional[int]
    created_at: str

@function()
def create_user(request: UserRequest) -> UserResponse:
    """Create a user with validation."""
    user_id = generate_user_id()

    return UserResponse(
        id=user_id,
        name=request.name,
        email=request.email,
        age=request.age,
        created_at=datetime.utcnow().isoformat()
    )

Testing Functions

Direct Testing

Test functions directly without the full platform:

import pytest
from agnt5.decorators import execute_component
from agnt5.components import ExecutionContext, ComponentType

def test_greet_function():
    # Test with execute_component
    result = execute_component(
        "greet_user",
        b'{"name": "Alice"}',
        context=None
    )

    # Result is JSON bytes
    import json
    parsed = json.loads(result.decode())
    assert parsed == "Hello, Alice!"

def test_context_function():
    # Create mock context
    ctx = ExecutionContext(
        invocation_id="test-123",
        component_type=ComponentType.FUNCTION
    )

    # Test directly
    result = context_aware(ctx, {"test": "data"})
    assert result["invocation_id"] == "test-123"

Async Testing

import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_function():
    result = await async_processor({"test": "data"})
    assert "original" in result
    assert "processed_at" in result

Mock Context Testing

from unittest.mock import Mock

def test_with_mock_context():
    # Create mock context
    mock_ctx = Mock(spec=ExecutionContext)
    mock_ctx.invocation_id = "mock-123"
    mock_ctx.component_type = ComponentType.FUNCTION
    mock_ctx.metadata = {"service_name": "test-service"}

    # Test function
    result = context_aware(mock_ctx, {"test": "data"})
    assert result["invocation_id"] == "mock-123"
    assert result["service_name"] == "test-service"

Function Registry

Registry Management

from agnt5.decorators import (
    get_registered_functions,
    clear_registry,
    get_function_metadata
)

# Get all registered functions
functions = get_registered_functions()

# Clear registry (useful for testing)
clear_registry()

# Re-register functions
@function()
def new_function(data: str) -> str:
    return data.upper()

# Inspect metadata
metadata = get_function_metadata(new_function)

Custom Registration

For advanced use cases, register functions manually:

from agnt5.decorators import register_function

def my_handler(data: str) -> str:
    return data.lower()

# Manual registration
register_function("custom_handler", my_handler)

Best Practices

Function Design

Keep functions focused - Each function should have a single responsibility
Use type hints - Improve documentation and enable validation
Handle errors gracefully - Return error information rather than raising exceptions
Log appropriately - Use structured logging for debugging and monitoring

Performance

Minimize imports - Import only what you need
Use async for I/O - Async functions for database queries and API calls
Cache expensive operations - Use local caching for repeated computations
Batch operations - Process multiple items together when possible

Testing

Test functions directly - Unit test without the platform
Mock external dependencies - Use mocks for databases, APIs, etc.
Test error conditions - Ensure error handling works correctly
Use fixtures - Share common test data and setup

Next Steps

Workflows - Multi-step orchestration patterns
Worker Runtime - Configure and deploy workers
API Reference - Complete decorator API reference
Examples - Real-world function examples