kiva-sdk 0.3.3

A multi-agent orchestration SDK for building intelligent workflows

Kiva SDK

⚠️ Important Notice: This project is currently in a rapid iteration/experimental phase, and the provided API may undergo disruptive changes at any time.

A multi-agent orchestration SDK for building intelligent workflows

Features

• Three Workflow Patterns: Router (simple), Supervisor (parallel), and Parliament (deliberative)
• Automatic Complexity Analysis: Intelligent workflow selection based on task complexity
• Parallel Agent Instances: Spawn multiple instances of the same agent for parallel subtask execution
• Modular Architecture: AgentRouter for organizing agents across multiple files
• Rich Console Output: Beautiful terminal visualization (optional)
• Error Recovery: Built-in error handling with recovery suggestions

Installation

uv add kiva-sdk

Setup & Configuration

Before running the SDK, you need to configure your API credentials using environment variables:

export KIVA_API_BASE="http://your-api-endpoint/v1"
export KIVA_API_KEY="your-api-key"
export KIVA_MODEL="your-model-name"

Quick Start

Basic Usage

from kiva import Kiva

kiva = Kiva(
    base_url="https://api.openai.com/v1",
    api_key="your-api-key",
    model="gpt-4o",
)

# Single-tool agent using decorator
@kiva.agent("weather", "Gets weather information")
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Sunny, 25°C in {city}"

# Multi-tool agent using class decorator
@kiva.agent("math", "Performs calculations")
class MathTools:
    def add(self, a: int, b: int) -> int:
        """Add two numbers."""
        return a + b
    
    def multiply(self, a: int, b: int) -> int:
        """Multiply two numbers."""
        return a * b

# Run with rich console output
kiva.run("What's the weather in Tokyo? Also calculate 15 * 8")

# Silent mode - no console output
result = kiva.run("What's the weather in Beijing?", console=False)
print(result)

Modular Application with AgentRouter

For larger applications, use AgentRouter to organize agents across multiple files:

# agents/weather.py
from kiva import AgentRouter

router = AgentRouter(prefix="weather")

@router.agent("forecast", "Gets weather forecasts")
def get_forecast(city: str) -> str:
    """Get weather forecast for a city."""
    return f"Sunny, 25°C in {city}"

# agents/math.py
from kiva import AgentRouter

router = AgentRouter(prefix="math")

@router.agent("calculator", "Performs calculations")
class Calculator:
    def add(self, a: int, b: int) -> int:
        """Add two numbers."""
        return a + b

# main.py
from kiva import Kiva
from agents.weather import router as weather_router
from agents.math import router as math_router

kiva = Kiva(base_url="...", api_key="...", model="gpt-4o")

kiva.include_router(weather_router)
kiva.include_router(math_router)

kiva.run("What's the weather in Tokyo? Calculate 15 * 8")

See AgentRouter Documentation for more details.

Workflow Patterns

Router Workflow

Routes tasks to a single most appropriate agent. Best for simple, single-domain queries.

Supervisor Workflow

Coordinates multiple agents executing in parallel. Supports spawning multiple instances of the same agent for parallel subtask processing. Ideal for multi-faceted tasks that can be decomposed into independent subtasks.

Parliament Workflow

Implements iterative deliberation with conflict resolution. Designed for complex reasoning tasks requiring consensus or validation.

Parallel Agent Instances

Kiva can spawn multiple instances of the same agent definition for parallel execution:

# The planner automatically decides when to use parallel instances
# For example, this task might spawn 3 instances of a search agent:
kiva.run("Search for information about AI, blockchain, and quantum computing")

Each instance has:

• Isolated context/scratchpad
• Independent execution
• Results aggregated automatically

See Parallel Instances Documentation for details.

Async Support

For async contexts, use run_async:

import asyncio
from kiva import Kiva

async def main():
    kiva = Kiva(base_url="...", api_key="...", model="gpt-4o")
    
    @kiva.agent("weather", "Gets weather")
    def get_weather(city: str) -> str:
        return f"{city}: Sunny"
    
    result = await kiva.run_async("Weather in Tokyo?", console=False)
    print(result)

asyncio.run(main())

Documentation

• AgentRouter - Modular Applications
• Parallel Agent Instances
• E2E Testing Guide

Testing

The SDK includes comprehensive unit and end-to-end tests:

# Unit tests (no API required)
uv run --dev pytest tests/ -v --ignore=tests/e2e/

# End-to-end tests (requires API configuration)
export KIVA_API_BASE="http://your-api-endpoint/v1"
export KIVA_API_KEY="your-api-key"
export KIVA_MODEL="your-model-name"

uv run --dev pytest tests/e2e/ -v

See E2E Testing Guide for more details.

License

MIT License