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.
Overview
The III SDK provides execution context that includes a logger and optional OpenTelemetry span. Context is automatically injected into function handlers and can be accessed via get_context().
Context
Execution context containing logger and tracing utilities.
from pydantic import BaseModel
class Context(BaseModel):
logger: Logger
trace: Any | None = None # OpenTelemetry span
Context-aware logger that emits OpenTelemetry LogRecords
The active OpenTelemetry span for adding custom attributes, events, etc.
Accessing Context
get_context
Get the current execution context within a function handler.
from iii import get_context
async def my_function(data):
ctx = get_context()
ctx.logger.info("Processing request", data={"user_id": data["user_id"]})
# Your logic here
ctx.logger.info("Request completed")
return {"status": "success"}
The current execution context, or a default context if none is set
Default Context
If no context is active, get_context() returns a default context with a basic logger:
ctx = get_context() # Always returns a Context, never None
ctx.logger.info("This works even outside function handlers")
Logger
Context-aware logger that emits OpenTelemetry LogRecords when OTel is active.
Log Methods
from iii import get_context
ctx = get_context()
# Info level
ctx.logger.info("User logged in", data={"user_id": "123"})
# Warning level
ctx.logger.warn("High memory usage", data={"usage_percent": 85})
# Error level
ctx.logger.error("Database connection failed", data={"error": "timeout"})
# Debug level
ctx.logger.debug("Request details", data={"headers": headers})
Additional structured data to include in the log record
Logger Behavior
When OpenTelemetry is initialized:
- Logs are emitted as OpenTelemetry LogRecords
- Includes trace context (span ID, trace ID)
- Exported to the III Engine
- Function name is automatically included
When OpenTelemetry is not initialized:
- Falls back to Python’s standard logging module
- Logs to console/file handlers as configured
- Still includes function name and data
Function Name
The logger automatically includes the function name:
iii.register_function("orders.process", process_order)
async def process_order(data):
ctx = get_context()
# Log includes function_name="orders.process"
ctx.logger.info("Order received")
with_context
Execute a function within a specific context (internal use).
from iii import with_context, Context, Logger
logger = Logger(function_name="my.function")
ctx = Context(logger=logger)
async def handler(ctx: Context) -> dict:
ctx.logger.info("Inside context")
return {"result": "success"}
result = await with_context(handler, ctx)
fn
Callable[[Context], Awaitable[T]]
required
Async function to execute with the context
The function’s return value
Example: Structured Logging
import asyncio
from iii import III, get_context, init_otel
init_otel() # Enable OpenTelemetry logging
iii = III("ws://localhost:49134")
async def create_user(data):
ctx = get_context()
ctx.logger.info("Creating user", data={
"email": data["email"],
"role": data.get("role", "user")
})
try:
# Validate email
if "@" not in data["email"]:
ctx.logger.warn("Invalid email format", data={"email": data["email"]})
raise ValueError("Invalid email")
# Save to database
user_id = save_user(data)
ctx.logger.info("User created successfully", data={
"user_id": user_id,
"email": data["email"]
})
return {"id": user_id, "email": data["email"]}
except Exception as e:
ctx.logger.error("Failed to create user", data={
"error": str(e),
"email": data.get("email")
})
raise
iii.register_function("users.create", create_user)
async def main():
await iii.connect()
try:
result = await iii.call("users.create", {
"email": "alice@example.com",
"role": "admin"
})
print(f"User created: {result}")
except Exception as e:
print(f"Failed: {e}")
if __name__ == "__main__":
asyncio.run(main())
Example: Logging with Trace Context
import asyncio
from iii import III, get_context, init_otel, get_tracer
from opentelemetry import trace
init_otel()
tracer = get_tracer()
iii = III("ws://localhost:49134")
async def process_order(data):
ctx = get_context()
with tracer.start_as_current_span("validate-order") as span:
ctx.logger.info("Validating order", data={"order_id": data["order_id"]})
# Validation logic
span.set_attribute("order.items", len(data.get("items", [])))
ctx.logger.info("Order validated")
with tracer.start_as_current_span("process-payment"):
ctx.logger.info("Processing payment", data={
"order_id": data["order_id"],
"amount": data["amount"]
})
# Payment logic
ctx.logger.info("Payment processed")
ctx.logger.info("Order completed", data={"order_id": data["order_id"]})
return {"status": "completed"}
iii.register_function("orders.process", process_order)
Example: Error Tracking
from iii import get_context
import traceback
async def risky_operation(data):
ctx = get_context()
try:
ctx.logger.debug("Starting risky operation", data={"input": data})
# Potentially failing operation
result = divide(data["a"], data["b"])
ctx.logger.debug("Operation succeeded", data={"result": result})
return {"result": result}
except ZeroDivisionError:
ctx.logger.error("Division by zero", data={
"a": data["a"],
"b": data["b"]
})
raise
except Exception as e:
ctx.logger.error("Unexpected error", data={
"error": str(e),
"traceback": traceback.format_exc()
})
raise
Example: Request/Response Logging
import time
from iii import get_context
async def api_handler(data):
ctx = get_context()
start_time = time.time()
# Log request
ctx.logger.info("API request received", data={
"method": data.get("method"),
"path": data.get("path"),
"user_id": data.get("user_id")
})
try:
# Process request
result = process_request(data)
# Log response
duration_ms = (time.time() - start_time) * 1000
ctx.logger.info("API request completed", data={
"status": "success",
"duration_ms": duration_ms
})
return result
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
ctx.logger.error("API request failed", data={
"status": "error",
"error": str(e),
"duration_ms": duration_ms
})
raise
Logger Implementation Details
The Logger class automatically:
- Detects OTel initialization: Uses
is_initialized() to check if OTel is active
- Emits LogRecords: When OTel is active, creates OpenTelemetry LogRecords with:
- Timestamp (nanoseconds)
- Severity level and text
- Message body
- Function name attribute
- Custom data attribute
- Trace context (span ID, trace ID, trace flags)
- Falls back to Python logging: When OTel is not active, uses Python’s logging module
- Severity mapping:
debug → DEBUG (5)info → INFO (9)warn → WARN (13)error → ERROR (17)
Best Practices
- Use structured data: Pass dictionaries to the
data parameter instead of formatting strings
- Include context: Add relevant identifiers (user_id, order_id, etc.) to logs
- Appropriate levels: Use info for business logic, debug for diagnostics, error for failures
- Avoid sensitive data: Don’t log passwords, tokens, or PII
- Measure performance: Log durations for slow operations
- Trace correlation: Logs automatically include trace context when OTel is active