Skip to content
SDK Documentation

Decorators API Reference

Decorators API Reference

The Klira SDK provides a unified set of decorators that automatically adapt to any LLM framework. These decorators add observability, tracing, and governance to your functions and classes.

Decorator Import

from klira.sdk.decorators import workflow, task, agent, tool, guardrails, mcp_guardrails

Core Decorators

@workflow

Decorator for high-level processes and workflows.

def workflow(
    name: Optional[str] = None,
    user_id: Optional[str] = None,
    conversation_id: Optional[str] = None,
    **kwargs: Any
) -> Callable[[Union[F, C]], Union[F, C]]

Parameters

ParameterTypeDefaultDescription
nameOptional[str]NoneName of the workflow. Defaults to function/class name
user_idOptional[str]NoneUser identifier for trace context
conversation_idOptional[str]NoneConversation identifier for trace context
**kwargsAny-Additional framework-specific arguments

Examples

Basic workflow:

@workflow(name="user_onboarding")
def onboard_user(user_data: dict) -> dict:
    return {"status": "success", "user_id": user_data["id"]}

Workflow with context:

@workflow(name="chat_session", user_id="user_123", conversation_id="conv_456")
async def chat_session(message: str) -> str:
    return await process_message(message)

Class-based workflow:

@workflow(name="data_pipeline")
class DataProcessingWorkflow:
    def __init__(self, config: dict):
        self.config = config


    def execute(self, data: list) -> list:
        return processed_data

@task

Decorator for individual operations and tasks.

def task(
    name: Optional[str] = None,
    **kwargs: Any
) -> Callable[[Union[F, C]], Union[F, C]]

Parameters

ParameterTypeDefaultDescription
nameOptional[str]NoneName of the task. Defaults to function/class name
**kwargsAny-Additional framework-specific arguments

Examples

Data processing task:

@task(name="extract_entities")
def extract_entities(text: str) -> list:
    return entities

Async task:

@task(name="fetch_user_data")
async def fetch_user_data(user_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"/users/{user_id}")
        return response.json()

@agent

Decorator for autonomous components and agents.

def agent(
    name: Optional[str] = None,
    agent_id: Optional[str] = None,
    **kwargs: Any
) -> Callable[[Union[F, C]], Union[F, C]]

Parameters

ParameterTypeDefaultDescription
nameOptional[str]NoneName of the agent. Defaults to function/class name
agent_idOptional[str]NoneSpecific agent identifier
**kwargsAny-Additional framework-specific arguments

Examples

OpenAI Agents SDK:

@agent(name="customer_service", agent_id="cs_001")
def create_customer_service_agent():
    return Agent(
        name="CustomerService",
        instructions="You are a helpful customer service assistant",
        tools=[get_order_status, process_refund]
    )

Custom agent class:

@agent(name="custom_agent")
class CustomAgent:
    def __init__(self, model: str):
        self.model = model


    async def process(self, input_text: str) -> str:
        return response

@tool

Decorator for utility functions and tools.

def tool(
    name: Optional[str] = None,
    agent_id: Optional[str] = None,
    tool_id: Optional[str] = None,
    fhir: bool = False,
    **kwargs: Any
) -> Callable[[Union[F, C]], Union[F, C]]

Parameters

ParameterTypeDefaultDescription
nameOptional[str]NoneName of the tool. Defaults to function/class name
agent_idOptional[str]NoneID of the agent using the tool
tool_idOptional[str]NoneSpecific tool identifier
fhirboolFalseMark this tool as FHIR-compliant for healthcare workflows
**kwargsAny-Additional framework-specific arguments

Examples

Simple tool:

@tool(name="calculator", tool_id="calc_001")
def calculate(expression: str) -> str:
    """Calculate a mathematical expression."""
    return str(eval(expression))

FHIR-compliant tool:

@tool(name="patient_lookup", fhir=True)
def lookup_patient(patient_id: str) -> dict:
    """Look up patient record via FHIR API."""
    return fhir_client.read("Patient", patient_id)

OpenAI function tool:

from agents import function_tool


@tool(name="weather_tool")
@function_tool()
def get_weather(location: str) -> str:
    """Get weather information for a location."""
    return f"Weather in {location}: Sunny, 72°F"

Governance Decorators

@guardrails

Decorator for applying policy enforcement and safety guardrails.

def guardrails(
    domain: Optional[str] = None,
    policy_set: Optional[str] = None,
    check_input: bool = True,
    check_output: bool = False,
) -> Callable[[F], F]

Parameters

ParameterTypeDefaultDescription
domainOptional[str]NonePolicy domain for routing (e.g., "healthcare", "finance")
policy_setOptional[str]NoneSpecific policy set to apply
check_inputboolTrueWhether to check function input against policies
check_outputboolFalseWhether to check function output against policies

_klira_guidelines Injection

When @guardrails is applied, the decorated function receives an additional keyword argument _klira_guidelines containing a list of policy guideline strings. Your function can use these to augment system prompts:

@workflow(name="chat")
@guardrails(domain="healthcare")
async def chat(message: str, _klira_guidelines: list[str] = None) -> str:
    system_prompt = "You are a helpful assistant."
    if _klira_guidelines:
        system_prompt += "\n\nGuidelines:\n" + "\n".join(_klira_guidelines)
    return await llm_call(system_prompt, message)

Examples

Basic guardrails:

@workflow(name="chat_workflow")
@guardrails()
def process_chat(user_input: str) -> str:
    return llm_response(user_input)

Domain-specific guardrails:

@workflow(name="clinical_chat")
@guardrails(domain="healthcare", check_output=True)
async def clinical_chat(message: str) -> str:
    return await clinical_llm_call(message)

Input and output checking:

@guardrails(check_input=True, check_output=True)
def sensitive_operation(input_data: str) -> str:
    return process_sensitive_data(input_data)

@mcp_guardrails

Decorator for MCP (Model Context Protocol) guardrails.

def mcp_guardrails(
    policy_set: Optional[PolicySet] = None,
    on_violation: str = ViolationMode.AUGMENT,
    conversation_id: Optional[str] = None,
    enabled: bool = True,
) -> Callable[[F], F]

Parameters

ParameterTypeDefaultDescription
policy_setOptional[PolicySet]NoneCustom policy set for evaluation
on_violationstrViolationMode.AUGMENTViolation handling mode: "block", "augment", or "warn"
conversation_idOptional[str]NoneConversation identifier for context
enabledboolTrueWhether guardrails are enabled

Example

from klira.sdk.decorators import mcp_guardrails, ViolationMode


@mcp_guardrails(on_violation=ViolationMode.BLOCK)
def get_user_data(user_id: str) -> str:
    """Function with MCP guardrail protection."""
    return f"User data for {user_id}"


@mcp_guardrails(on_violation=ViolationMode.AUGMENT)
async def process_message(message: str) -> str:
    """Async function with augmentation on violations."""
    return await llm_process(message)

Decorator Composition

Combining Decorators

@workflow(name="secure_chat", user_id="user_123", conversation_id="conv_456")
@guardrails(domain="finance", check_input=True, check_output=True)
def secure_chat_workflow(user_input: str) -> str:
    """Secure chat workflow with full observability and governance."""
    return process_chat_securely(user_input)
  1. Framework decorators (@workflow, @task, @agent, @tool) — outermost
  2. Governance decorators (@guardrails) — middle
  3. Framework-specific decorators (e.g., @function_tool) — innermost
@workflow(name="example")     # 1. Framework decorator (outermost)
@guardrails()                 # 2. Governance decorator
@function_tool()              # 3. Framework-specific (innermost)
def example_function():
    pass

Framework Detection

All decorators automatically detect the framework being used:

  • OpenAI Agents SDK: Automatic detection of Agent, Runner, function_tool
  • LangChain: Automatic detection of agents, chains, and tools
  • LlamaIndex: Automatic detection of query engines and chat engines
  • Custom Frameworks: Plugin-based detection for custom implementations

Error Handling

All decorators include robust error handling. Functions execute even if decorators fail — errors are logged but don’t break the application.

Async Support

All decorators support both synchronous and asynchronous functions:

@workflow(name="async_workflow")
@guardrails()
async def async_workflow(data: str) -> str:
    result = await async_llm_call(data)
    return result

Best Practices

Use Descriptive Names

# Good — descriptive names
@workflow(name="user_onboarding_workflow")
@task(name="validate_user_email")
@tool(name="email_validator")

Apply Guardrails Based on Sensitivity

# High-sensitivity function
@guardrails(domain="healthcare", check_input=True, check_output=True)
def process_clinical_data(data: str) -> str:
    pass


# Low-sensitivity function
@guardrails(check_output=False)
def general_chat(message: str) -> str:
    pass