> For the complete documentation index, see [llms.txt](/llms.txt).
> A full single-fetch corpus is available at [llms-full.txt](/llms-full.txt).
---
last_verified: 2026-05-13
title: Functions
description: Handler decorators and function execution in the AGNT5 Python SDK
category: 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

```python
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:

```python
@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:

```python
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. |

```python
# 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:

```python
@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:

```python
@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

```python
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

```python
@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

```python
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

```python
@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

```python
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:**

```python
{
    "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:

```python
@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

```python
@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

```python
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

```python
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:

```python
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:

```python
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

```python
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

```python
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

```python
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:

```python
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

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

### Performance

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

### Testing

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

## Next Steps

- [Workflows](workflows) - Multi-step orchestration patterns
- [Worker Runtime](worker) - Configure and deploy workers
- [API Reference](api/decorators) - Complete decorator API reference
- [Examples](examples/basic-worker) - Real-world function examples