ADK Multi-Agent Architecture

Serverless AI Agents at Your Service

🚀 Serverless Agents with Function Calling

ADK exposes Kubiya’s serverless AI agents that leverage state-of-the-art techniques:

Function Calling: Agents use simple function calls to execute workflow tools
Serverless Execution: Spin up on-demand, no persistent processes
Tool Integration: Seamlessly call workflow steps as functions
Best-in-Class Models: Leverages top AI models for reasoning
Deterministic Outputs: Structured responses that fit into workflows

The Power of Inline Agents

Unlike traditional agent frameworks that require complex setups, Kubiya’s inline agents are:

Just workflow steps - No separate agent infrastructure
Function-based - Agents call tools through simple function interfaces
Context-aware - Access workflow state and previous outputs
Serverless - No always-on processes consuming resources
Integrated - Part of your workflow DAG, not external

# Serverless agent as a workflow step
.step("intelligent-analysis")
.inline_agent(
    message="Analyze system health and recommend actions",
    agent_name="sre-assistant",
    ai_instructions="You are an SRE expert. Use the provided tools to gather metrics.",
    
    # Agent uses function calling to execute tools
    tools=[
        {
            "name": "get_cpu_usage",
            "type": "function",
            "description": "Get current CPU usage across pods",
            "implementation": "kubectl top pods | awk '{print $2}'"
        },
        {
            "name": "check_error_logs", 
            "type": "function",
            "description": "Count errors in application logs",
            "implementation": "kubectl logs -l app=myapp | grep ERROR | wc -l"
        }
    ],
    
    runners=["kubiya-hosted"]  # Serverless execution
)
.output("SRE_ANALYSIS")

How ADK Agents Work

The ADK provider orchestrates multiple specialized agents to transform natural language into production-ready workflows:

Agent Roles

1. Loop Orchestrator

The Loop Orchestrator is the conductor of the entire process:

Purpose: Coordinates all other agents
Responsibilities:
- Manages the generation lifecycle
- Handles error recovery and retries
- Enforces iteration limits
- Maintains conversation state

# Orchestrator configuration
config = ADKConfig(
    max_loop_iterations=3,  # Maximum refinement attempts
    enable_streaming=True,  # Stream agent outputs
    timeout=300            # Overall timeout
)

2. Context Loader Agent

Loads platform-specific context to inform workflow generation:

Fetches: Runners, integrations, secrets, organization settings
Outputs: Structured context for the generator
Uses: Kubiya API tools

The Context Loader ensures generated workflows use only available resources and integrations.

Loaded Context Includes:

Available runners and their capabilities
Installed integrations (Slack, AWS, etc.)
Accessible secrets (names only, not values)
Organization policies and constraints

3. Workflow Generator Agent

The creative powerhouse that generates workflow code:

Model: DeepSeek V3 (default) or configurable
Input: Task description + platform context
Output: Python SDK code using the DSL

from kubiya_workflow_sdk.dsl import Workflow, Step, parallel

workflow = Workflow(name="backup-databases")
workflow.description = "Automated database backup workflow"
workflow.runner = "kubiya-hosted"

# Parallel backup of multiple databases
backup_steps = []
for db in ["users", "orders", "inventory"]:
    step = Step(
        name=f"backup_{db}",
        tool="pg_dump",
        parameters={
            "database": db,
            "output": f"/backups/{db}_{{date}}.sql"
        }
    )
    backup_steps.append(step)

workflow.add_step(parallel(*backup_steps))

# Compress backups
workflow.add_step(Step(
    name="compress",
    tool="tar",
    parameters={
        "input": "/backups/*.sql",
        "output": "/backups/backup_{{date}}.tar.gz"
    }
))

from kubiya_workflow_sdk.dsl import Workflow, Step, parallel

workflow = Workflow(name="backup-databases")
workflow.description = "Automated database backup workflow"
workflow.runner = "kubiya-hosted"

# Parallel backup of multiple databases
backup_steps = []
for db in ["users", "orders", "inventory"]:
    step = Step(
        name=f"backup_{db}",
        tool="pg_dump",
        parameters={
            "database": db,
            "output": f"/backups/{db}_{{date}}.sql"
        }
    )
    backup_steps.append(step)

workflow.add_step(parallel(*backup_steps))

# Compress backups
workflow.add_step(Step(
    name="compress",
    tool="tar",
    parameters={
        "input": "/backups/*.sql",
        "output": "/backups/backup_{{date}}.tar.gz"
    }
))

The generator uses sophisticated prompts:

Platform capabilities and constraints
Best practices for the SDK
Security considerations
Performance optimization hints

4. Compiler Agent

Validates and compiles the generated code:

Validates: Syntax, imports, workflow structure
Compiles: Python code to workflow JSON
Checks: Dependencies, circular references
Reports: Errors and warnings

Common Validation Checks:

✓ Valid Python syntax
✓ Correct SDK imports and usage
✓ Runner exists and is accessible
✓ Required integrations are available
✓ Step dependencies are valid
✓ No circular dependencies

Fixes errors through intelligent iteration:

Input: Original code + compilation errors
Process: Analyzes errors and generates fixes
Smart: Learns from previous attempts

The refinement agent will attempt up to max_loop_iterations fixes before giving up.

Refinement Strategies:

Fix syntax errors
Resolve missing imports
Correct API usage
Add error handling
Optimize performance

6. Output Agent

Formats the final result:

Formats: JSON, YAML, or streaming events
Adds: Metadata, usage instructions
Handles: Both plan and act modes

Agent Communication

Agents communicate through structured messages:

# Example agent message
message = {
    "role": "workflow_generator",
    "content": {
        "generated_code": "...",
        "confidence": 0.95,
        "alternatives": []
    },
    "metadata": {
        "model": "deepseek-v3",
        "tokens_used": 1234,
        "generation_time": 2.5
    }
}

Customizing Agent Behavior

Model Selection

Different models for different agents:

config = ADKConfig(
    model_overrides={
        "orchestrator": "together_ai/deepseek-ai/DeepSeek-V3",
        "workflow_generator": "together_ai/Qwen/QwQ-32B-Preview",
        "compiler": "together_ai/google/gemma-2b-it",
        "refinement": "together_ai/deepseek-ai/DeepSeek-V3"
    }
)

Custom Tools

Add custom tools to agents:

from kubiya_workflow_sdk.providers.adk.tools import Tool

# Custom validation tool
@Tool(name="custom_validator")
def validate_security_policies(workflow_code: str) -> dict:
    """Validate workflow against security policies."""
    # Custom validation logic
    return {"valid": True, "warnings": []}

# Add to compiler agent
compiler_agent.add_tool(validate_security_policies)

Agent Hooks

Intercept agent communications:

def on_agent_message(agent_name: str, message: dict):
    """Log or modify agent messages."""
    print(f"[{agent_name}] {message.get('content', {}).get('status')}")
    
    # Modify message if needed
    if agent_name == "workflow_generator":
        message["metadata"]["custom_flag"] = True
    
    return message

adk.set_message_hook(on_agent_message)

Performance Optimization

Caching

Parallel Processing

Early Stopping

Monitoring Agents

Logging

Enable detailed agent logging:

import logging

# Enable ADK agent logging
logging.getLogger("kubiya_workflow_sdk.providers.adk.agents").setLevel(logging.DEBUG)

# Or specific agents
logging.getLogger("kubiya_workflow_sdk.providers.adk.agents.generator").setLevel(logging.DEBUG)

Metrics

Track agent performance:

# Get agent metrics
metrics = adk.get_agent_metrics()

print(f"Total requests: {metrics['total_requests']}")
print(f"Average generation time: {metrics['avg_generation_time']}s")
print(f"Refinement rate: {metrics['refinement_rate']}%")
print(f"Success rate: {metrics['success_rate']}%")

Streaming Agent Events

Monitor agents in real-time:

async for event in adk.compose(task="...", stream=True, include_agent_events=True):
    if event["type"] == "agent_event":
        agent = event["agent"]
        status = event["status"]
        print(f"[{agent}] {status}")

Best Practices

Model Selection

Use powerful models for generation, fast models for validation

Context Optimization

Provide only necessary context to reduce token usage

Error Handling

Always set reasonable iteration limits and timeouts

Monitoring

Track agent metrics to optimize performance

Troubleshooting

Agent Timeouts

Refinement Loops

Context Errors

If context loading fails:

# Check API connectivity
client = KubiyaClient()
print(client.get_organization_info())

# Provide manual context
result = await adk.compose(
    task="...",
    context={
        "runners": ["kubiya-hosted"],
        "integrations": ["slack", "github"]
    }
)

Next Steps

Streaming

Learn about streaming responses

Custom Agents

Build your own agents

API Reference

Complete API documentation

Examples

More complex examples

On this page

ADK Multi-Agent Architecture
Serverless AI Agents at Your Service
The Power of Inline Agents
How ADK Agents Work
Agent Roles
1. Loop Orchestrator
2. Context Loader Agent
3. Workflow Generator Agent
4. Compiler Agent
5. Refinement Agent
6. Output Agent
Agent Communication
Customizing Agent Behavior
Model Selection
Custom Tools
Agent Hooks
Performance Optimization
Monitoring Agents
Logging
Metrics
Streaming Agent Events
Best Practices
Troubleshooting
Next Steps

ADK Multi-Agent Architecture

Serverless AI Agents at Your Service

🚀 Serverless Agents with Function Calling

ADK exposes Kubiya’s serverless AI agents that leverage state-of-the-art techniques:

Function Calling: Agents use simple function calls to execute workflow tools
Serverless Execution: Spin up on-demand, no persistent processes
Tool Integration: Seamlessly call workflow steps as functions
Best-in-Class Models: Leverages top AI models for reasoning
Deterministic Outputs: Structured responses that fit into workflows

The Power of Inline Agents

Unlike traditional agent frameworks that require complex setups, Kubiya’s inline agents are:

Just workflow steps - No separate agent infrastructure
Function-based - Agents call tools through simple function interfaces
Context-aware - Access workflow state and previous outputs
Serverless - No always-on processes consuming resources
Integrated - Part of your workflow DAG, not external

# Serverless agent as a workflow step
.step("intelligent-analysis")
.inline_agent(
    message="Analyze system health and recommend actions",
    agent_name="sre-assistant",
    ai_instructions="You are an SRE expert. Use the provided tools to gather metrics.",
    
    # Agent uses function calling to execute tools
    tools=[
        {
            "name": "get_cpu_usage",
            "type": "function",
            "description": "Get current CPU usage across pods",
            "implementation": "kubectl top pods | awk '{print $2}'"
        },
        {
            "name": "check_error_logs", 
            "type": "function",
            "description": "Count errors in application logs",
            "implementation": "kubectl logs -l app=myapp | grep ERROR | wc -l"
        }
    ],
    
    runners=["kubiya-hosted"]  # Serverless execution
)
.output("SRE_ANALYSIS")

How ADK Agents Work

The ADK provider orchestrates multiple specialized agents to transform natural language into production-ready workflows:

Agent Roles

1. Loop Orchestrator

The Loop Orchestrator is the conductor of the entire process:

Purpose: Coordinates all other agents
Responsibilities:
- Manages the generation lifecycle
- Handles error recovery and retries
- Enforces iteration limits
- Maintains conversation state

# Orchestrator configuration
config = ADKConfig(
    max_loop_iterations=3,  # Maximum refinement attempts
    enable_streaming=True,  # Stream agent outputs
    timeout=300            # Overall timeout
)

2. Context Loader Agent

Loads platform-specific context to inform workflow generation:

Fetches: Runners, integrations, secrets, organization settings
Outputs: Structured context for the generator
Uses: Kubiya API tools

The Context Loader ensures generated workflows use only available resources and integrations.

Loaded Context Includes:

Available runners and their capabilities
Installed integrations (Slack, AWS, etc.)
Accessible secrets (names only, not values)
Organization policies and constraints

3. Workflow Generator Agent

The creative powerhouse that generates workflow code:

Model: DeepSeek V3 (default) or configurable
Input: Task description + platform context
Output: Python SDK code using the DSL

from kubiya_workflow_sdk.dsl import Workflow, Step, parallel

workflow = Workflow(name="backup-databases")
workflow.description = "Automated database backup workflow"
workflow.runner = "kubiya-hosted"

# Parallel backup of multiple databases
backup_steps = []
for db in ["users", "orders", "inventory"]:
    step = Step(
        name=f"backup_{db}",
        tool="pg_dump",
        parameters={
            "database": db,
            "output": f"/backups/{db}_{{date}}.sql"
        }
    )
    backup_steps.append(step)

workflow.add_step(parallel(*backup_steps))

# Compress backups
workflow.add_step(Step(
    name="compress",
    tool="tar",
    parameters={
        "input": "/backups/*.sql",
        "output": "/backups/backup_{{date}}.tar.gz"
    }
))

from kubiya_workflow_sdk.dsl import Workflow, Step, parallel

workflow = Workflow(name="backup-databases")
workflow.description = "Automated database backup workflow"
workflow.runner = "kubiya-hosted"

# Parallel backup of multiple databases
backup_steps = []
for db in ["users", "orders", "inventory"]:
    step = Step(
        name=f"backup_{db}",
        tool="pg_dump",
        parameters={
            "database": db,
            "output": f"/backups/{db}_{{date}}.sql"
        }
    )
    backup_steps.append(step)

workflow.add_step(parallel(*backup_steps))

# Compress backups
workflow.add_step(Step(
    name="compress",
    tool="tar",
    parameters={
        "input": "/backups/*.sql",
        "output": "/backups/backup_{{date}}.tar.gz"
    }
))

The generator uses sophisticated prompts:

Platform capabilities and constraints
Best practices for the SDK
Security considerations
Performance optimization hints

4. Compiler Agent

Validates and compiles the generated code:

Validates: Syntax, imports, workflow structure
Compiles: Python code to workflow JSON
Checks: Dependencies, circular references
Reports: Errors and warnings

Common Validation Checks:

✓ Valid Python syntax
✓ Correct SDK imports and usage
✓ Runner exists and is accessible
✓ Required integrations are available
✓ Step dependencies are valid
✓ No circular dependencies

Fixes errors through intelligent iteration:

Input: Original code + compilation errors
Process: Analyzes errors and generates fixes
Smart: Learns from previous attempts

The refinement agent will attempt up to max_loop_iterations fixes before giving up.

Refinement Strategies:

Fix syntax errors
Resolve missing imports
Correct API usage
Add error handling
Optimize performance

6. Output Agent

Formats the final result:

Formats: JSON, YAML, or streaming events
Adds: Metadata, usage instructions
Handles: Both plan and act modes

Agent Communication

Agents communicate through structured messages:

# Example agent message
message = {
    "role": "workflow_generator",
    "content": {
        "generated_code": "...",
        "confidence": 0.95,
        "alternatives": []
    },
    "metadata": {
        "model": "deepseek-v3",
        "tokens_used": 1234,
        "generation_time": 2.5
    }
}

Customizing Agent Behavior

Model Selection

Different models for different agents:

config = ADKConfig(
    model_overrides={
        "orchestrator": "together_ai/deepseek-ai/DeepSeek-V3",
        "workflow_generator": "together_ai/Qwen/QwQ-32B-Preview",
        "compiler": "together_ai/google/gemma-2b-it",
        "refinement": "together_ai/deepseek-ai/DeepSeek-V3"
    }
)

Custom Tools

Add custom tools to agents:

from kubiya_workflow_sdk.providers.adk.tools import Tool

# Custom validation tool
@Tool(name="custom_validator")
def validate_security_policies(workflow_code: str) -> dict:
    """Validate workflow against security policies."""
    # Custom validation logic
    return {"valid": True, "warnings": []}

# Add to compiler agent
compiler_agent.add_tool(validate_security_policies)

Agent Hooks

Intercept agent communications:

def on_agent_message(agent_name: str, message: dict):
    """Log or modify agent messages."""
    print(f"[{agent_name}] {message.get('content', {}).get('status')}")
    
    # Modify message if needed
    if agent_name == "workflow_generator":
        message["metadata"]["custom_flag"] = True
    
    return message

adk.set_message_hook(on_agent_message)

Performance Optimization

Caching

Parallel Processing

Early Stopping

Monitoring Agents

Logging

Enable detailed agent logging:

import logging

# Enable ADK agent logging
logging.getLogger("kubiya_workflow_sdk.providers.adk.agents").setLevel(logging.DEBUG)

# Or specific agents
logging.getLogger("kubiya_workflow_sdk.providers.adk.agents.generator").setLevel(logging.DEBUG)

Metrics

Track agent performance:

# Get agent metrics
metrics = adk.get_agent_metrics()

print(f"Total requests: {metrics['total_requests']}")
print(f"Average generation time: {metrics['avg_generation_time']}s")
print(f"Refinement rate: {metrics['refinement_rate']}%")
print(f"Success rate: {metrics['success_rate']}%")

Streaming Agent Events

Monitor agents in real-time:

async for event in adk.compose(task="...", stream=True, include_agent_events=True):
    if event["type"] == "agent_event":
        agent = event["agent"]
        status = event["status"]
        print(f"[{agent}] {status}")

Best Practices

Model Selection

Use powerful models for generation, fast models for validation

Context Optimization

Provide only necessary context to reduce token usage

Error Handling

Always set reasonable iteration limits and timeouts

Monitoring

Track agent metrics to optimize performance

Troubleshooting

Agent Timeouts

Refinement Loops

Context Errors

If context loading fails:

# Check API connectivity
client = KubiyaClient()
print(client.get_organization_info())

# Provide manual context
result = await adk.compose(
    task="...",
    context={
        "runners": ["kubiya-hosted"],
        "integrations": ["slack", "github"]
    }
)

Next Steps

Streaming

Learn about streaming responses

Custom Agents

Build your own agents

API Reference

Complete API documentation

Examples

More complex examples

On this page

ADK Multi-Agent Architecture
Serverless AI Agents at Your Service
The Power of Inline Agents
How ADK Agents Work
Agent Roles
1. Loop Orchestrator
2. Context Loader Agent
3. Workflow Generator Agent
4. Compiler Agent
5. Refinement Agent
6. Output Agent
Agent Communication
Customizing Agent Behavior
Model Selection
Custom Tools
Agent Hooks
Performance Optimization
Monitoring Agents
Logging
Metrics
Streaming Agent Events
Best Practices
Troubleshooting
Next Steps

​ADK Multi-Agent Architecture

​Serverless AI Agents at Your Service

🚀 Serverless Agents with Function Calling

​The Power of Inline Agents

​How ADK Agents Work

​Agent Roles

​1. Loop Orchestrator

​2. Context Loader Agent

​3. Workflow Generator Agent

​4. Compiler Agent

​5. Refinement Agent

​6. Output Agent

​Agent Communication

​Customizing Agent Behavior

​Model Selection

​Custom Tools

​Agent Hooks

​Performance Optimization

​Monitoring Agents

​Logging

​Metrics

​Streaming Agent Events

​Best Practices

Model Selection

Context Optimization

Error Handling

Monitoring

​Troubleshooting

​Next Steps

Streaming

Custom Agents

API Reference

Examples

SDK Reference

​ADK Multi-Agent Architecture

​Serverless AI Agents at Your Service

🚀 Serverless Agents with Function Calling

​The Power of Inline Agents

​How ADK Agents Work

​Agent Roles

​1. Loop Orchestrator

​2. Context Loader Agent

​3. Workflow Generator Agent

​4. Compiler Agent

​5. Refinement Agent

​6. Output Agent

​Agent Communication

​Customizing Agent Behavior

​Model Selection

​Custom Tools

​Agent Hooks

​Performance Optimization

​Monitoring Agents

​Logging

​Metrics

​Streaming Agent Events

​Best Practices

Model Selection

Context Optimization

Error Handling

Monitoring

​Troubleshooting

​Next Steps

Streaming

Custom Agents

API Reference

Examples

ADK Multi-Agent Architecture

Serverless AI Agents at Your Service

The Power of Inline Agents

How ADK Agents Work

Agent Roles

1. Loop Orchestrator

2. Context Loader Agent

3. Workflow Generator Agent

4. Compiler Agent

5. Refinement Agent

6. Output Agent

Agent Communication

Customizing Agent Behavior

Model Selection

Custom Tools

Agent Hooks

Performance Optimization

Monitoring Agents

Logging

Metrics

Streaming Agent Events

Best Practices

Troubleshooting

Next Steps

ADK Multi-Agent Architecture

Serverless AI Agents at Your Service

The Power of Inline Agents

How ADK Agents Work

Agent Roles

1. Loop Orchestrator

2. Context Loader Agent

3. Workflow Generator Agent

4. Compiler Agent

5. Refinement Agent

6. Output Agent

Agent Communication

Customizing Agent Behavior

Model Selection

Custom Tools

Agent Hooks

Performance Optimization

Monitoring Agents

Logging

Metrics

Streaming Agent Events

Best Practices

Troubleshooting

Next Steps