> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
> Use this file to discover all available pages before exploring further.

# Functions

> Register and invoke functions in the III SDK

## Function Registration

### register\_function

Register a function handler that can be invoked by other workers.

```python theme={null}
async def greet(data):
    name = data.get("name", "World")
    return {"message": f"Hello, {name}!"}

ref = iii.register_function("greet", greet)
```

<ParamField path="path" type="str" required>
  Unique function ID (e.g., `"users.create"`, `"orders.process"`)
</ParamField>

<ParamField path="handler" type="Callable[[Any], Awaitable[Any]]" required>
  Async function that receives invocation data and returns a result
</ParamField>

<ParamField path="description" type="str">
  Human-readable description of what the function does
</ParamField>

<ParamField path="metadata" type="dict[str, Any]">
  Additional metadata (e.g., `{"version": "1.0", "team": "platform"}`)
</ParamField>

<ResponseField name="ref" type="FunctionRef">
  Reference object with `id` and `unregister()` method
</ResponseField>

### Handler Signature

Function handlers must be async functions:

```python theme={null}
from typing import Any

async def my_handler(data: Any) -> Any:
    """Process the input data and return a result."""
    # Your logic here
    return result
```

The `data` parameter contains the invocation payload. Return values are automatically serialized.

### Unregistering Functions

```python theme={null}
# Option 1: Using the returned reference
ref = iii.register_function("my.function", handler)
ref.unregister()

# Option 2: Using the reference ID
ref.id  # "my.function"
```

## HTTP Functions

### register\_http\_function

Register a function that invokes an external HTTP endpoint instead of a Python handler.

```python theme={null}
from iii import HttpInvocationConfig, HttpAuthBearer

config = HttpInvocationConfig(
    url="https://api.example.com/process",
    method="POST",
    timeout_ms=30000,
    headers={"Content-Type": "application/json"},
    auth=HttpAuthBearer(token_key="BEARER_TOKEN_ENV_VAR")
)

ref = iii.register_http_function("external.process", config)
```

<ParamField path="id" type="str" required>
  Unique function ID
</ParamField>

<ParamField path="config" type="HttpInvocationConfig" required>
  HTTP invocation configuration
</ParamField>

<ResponseField name="ref" type="FunctionRef">
  Reference object with `id` and `unregister()` method
</ResponseField>

### HttpInvocationConfig

```python theme={null}
class HttpInvocationConfig(BaseModel):
    url: str
    method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "POST"
    timeout_ms: int | None = None
    headers: dict[str, str] | None = None
    auth: HttpAuthConfig | None = None
```

<ParamField path="url" type="str" required>
  The HTTP endpoint URL
</ParamField>

<ParamField path="method" type="str" default="POST">
  HTTP method: `"GET"`, `"POST"`, `"PUT"`, `"PATCH"`, or `"DELETE"`
</ParamField>

<ParamField path="timeout_ms" type="int">
  Request timeout in milliseconds
</ParamField>

<ParamField path="headers" type="dict[str, str]">
  HTTP headers to include in the request
</ParamField>

<ParamField path="auth" type="HttpAuthConfig">
  Authentication configuration (HMAC, Bearer, or API Key)
</ParamField>

### HTTP Authentication

#### Bearer Token

```python theme={null}
from iii import HttpAuthBearer

auth = HttpAuthBearer(token_key="MY_TOKEN_ENV_VAR")
```

<ParamField path="token_key" type="str" required>
  Environment variable name containing the bearer token
</ParamField>

#### API Key

```python theme={null}
from iii import HttpAuthApiKey

auth = HttpAuthApiKey(
    header="X-API-Key",
    value_key="API_KEY_ENV_VAR"
)
```

<ParamField path="header" type="str" required>
  HTTP header name for the API key
</ParamField>

<ParamField path="value_key" type="str" required>
  Environment variable name containing the API key
</ParamField>

#### HMAC

```python theme={null}
from iii import HttpAuthHmac

auth = HttpAuthHmac(secret_key="HMAC_SECRET_ENV_VAR")
```

<ParamField path="secret_key" type="str" required>
  Environment variable name containing the HMAC secret
</ParamField>

## Function Invocation

### call

Invoke a function and await the response.

```python theme={null}
result = await iii.call("orders.process", {"order_id": "123"})
print(result)  # {"status": "processed"}
```

<ParamField path="path" type="str" required>
  The function ID to invoke
</ParamField>

<ParamField path="data" type="Any" required>
  Data to pass to the function
</ParamField>

<ParamField path="timeout" type="float" default={30.0}>
  Timeout in seconds
</ParamField>

<ResponseField name="result" type="Any">
  The function's return value
</ResponseField>

### call\_void

Invoke a function without waiting for a response (fire-and-forget).

```python theme={null}
iii.call_void("notifications.send", {"user_id": "456", "message": "Hello"})
```

<ParamField path="path" type="str" required>
  The function ID to invoke
</ParamField>

<ParamField path="data" type="Any" required>
  Data to pass to the function
</ParamField>

### trigger / trigger\_void

Aliases for `call` and `call_void`:

```python theme={null}
# Same as call()
result = await iii.trigger("my.function", data)

# Same as call_void()
iii.trigger_void("my.function", data)
```

## Error Handling

Functions can raise exceptions, which are propagated to the caller:

```python theme={null}
async def divide(data):
    a = data["a"]
    b = data["b"]
    if b == 0:
        raise ValueError("Division by zero")
    return {"result": a / b}

iii.register_function("math.divide", divide)

# Calling with invalid data
try:
    result = await iii.call("math.divide", {"a": 10, "b": 0})
except Exception as e:
    print(f"Error: {e}")  # "Error: Division by zero"
```

## Example: Service Architecture

```python theme={null}
import asyncio
from iii import III, get_context

iii = III("ws://localhost:49134")

# User service
async def create_user(data):
    ctx = get_context()
    ctx.logger.info("Creating user", data={"email": data["email"]})
    
    # Save to database...
    user_id = "user123"
    
    # Trigger welcome email
    iii.call_void("email.send_welcome", {"user_id": user_id})
    
    return {"id": user_id, "email": data["email"]}

# Email service
async def send_welcome_email(data):
    ctx = get_context()
    user_id = data["user_id"]
    
    # Fetch user details
    user = await iii.call("users.get", {"id": user_id})
    
    ctx.logger.info(f"Sending welcome email to {user['email']}")
    # Send email...
    
    return {"status": "sent"}

iii.register_function(
    "users.create",
    create_user,
    description="Create a new user account",
    metadata={"service": "users", "version": "1.0"}
)

iii.register_function(
    "email.send_welcome",
    send_welcome_email,
    description="Send welcome email to new users",
    metadata={"service": "email"}
)

async def main():
    await iii.connect()
    
    # Test the flow
    user = await iii.call("users.create", {"email": "alice@example.com"})
    print(f"Created user: {user}")
    
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())
```