API Reference

Module Exports

from kapso.deployment import (
    # Main factory
    DeploymentFactory,
    
    # Enums and configs
    DeployStrategy,
    DeployConfig,
    
    # Data classes
    DeploymentSetting,
    AdaptationResult,
    DeploymentInfo,
    
    # Base classes
    Software,
)

DeploymentFactory

Factory for creating deployed Software instances.

Methods

create()

Create a deployed Software instance.

@classmethod
def create(
    cls,
    strategy: DeployStrategy,
    config: DeployConfig,
    strategies: Optional[List[str]] = None,
) -> Software

Parameters:

Parameter	Type	Description
`strategy`	`DeployStrategy`	Deployment target (AUTO, LOCAL, DOCKER, etc.)
`config`	`DeployConfig`	Deployment configuration
`strategies`	`List[str]`	Optional list of allowed strategies (for AUTO)

Returns: Software instance with unified interface Raises:

ValueError if strategy not in allowed list
RuntimeError if adaptation fails

Example:

from kapso.deployment import DeploymentFactory, DeployStrategy, DeployConfig

config = DeployConfig(
    solution=my_solution,
    env_vars={"API_KEY": "xxx"},
    timeout=300,
)

# Auto-select best strategy
deployed_program = DeploymentFactory.create(DeployStrategy.AUTO, config)

# Force specific strategy
deployed_program = DeploymentFactory.create(DeployStrategy.MODAL, config)

# Restrict to certain strategies
deployed_program = DeploymentFactory.create(
    DeployStrategy.AUTO, 
    config,
    strategies=["local", "docker"]
)

list_strategies()

List all available deployment strategies.

@classmethod
def list_strategies(cls) -> List[str]

Returns: List of strategy names Example:

strategies = DeploymentFactory.list_strategies()
print(strategies)
# ['bentoml', 'docker', 'langgraph', 'local', 'modal']

explain_strategy()

Get explanation of a deployment strategy.

@classmethod
def explain_strategy(cls, strategy: str) -> str

Parameters:

Parameter	Type	Description
`strategy`	`str`	Strategy name

Returns: One-line description from selector instruction Example:

desc = DeploymentFactory.explain_strategy("modal")
print(desc)
# "Serverless GPU deployment on Modal.com with auto-scaling."

print_strategies_info()

Print information about all deployment strategies.

@classmethod
def print_strategies_info(cls) -> None

Example:

DeploymentFactory.print_strategies_info()
# Available Deployment Strategies:
# ==================================================
#   bentoml: Production ML service deployment...
#   docker: Run in an isolated Docker container...
#   ...

DeployStrategy

Enum of available deployment strategies. Auto-generated from strategies/ directory.

from kapso.deployment import DeployStrategy

# Always available
DeployStrategy.AUTO      # Let system choose

# Discovered from strategies/
DeployStrategy.LOCAL     # Local Python process
DeployStrategy.DOCKER    # Docker container
DeployStrategy.MODAL     # Modal.com serverless
DeployStrategy.BENTOML   # BentoML/BentoCloud
DeployStrategy.LANGGRAPH # LangGraph Platform

Usage:

# Use enum value
strategy = DeployStrategy.MODAL

# Get string value
print(strategy.value)  # "modal"

# Compare
if strategy == DeployStrategy.AUTO:
    print("Auto-selecting...")

DeployConfig

Configuration for deploying software.

@dataclass
class DeployConfig:
    solution: SolutionResult
    env_vars: Dict[str, str] = None
    timeout: int = 300
    coding_agent: str = "claude_code"

Fields:

Field	Type	Default	Description
`solution`	`SolutionResult`	Required	The built solution from `Kapso.evolve()`
`env_vars`	`Dict[str, str]`	`{}`	Environment variables
`timeout`	`int`	`300`	Execution timeout in seconds
`coding_agent`	`str`	`"claude_code"`	Coding agent for adaptation

Properties:

Property	Type	Description
`code_path`	`str`	Path to the generated code
`goal`	`str`	The original goal/objective

Example:

from kapso.deployment import DeployConfig

config = DeployConfig(
    solution=solution,
    env_vars={
        "API_KEY": "xxx",
        "DEBUG": "true",
    },
    timeout=600,
    coding_agent="gemini",
)

print(config.code_path)  # "/path/to/solution"
print(config.goal)       # "Create a sentiment API"

Software (Abstract Base Class)

Unified interface for deployed software. Users interact with this class.

Methods

run()

Execute the software with given inputs.

@abstractmethod
def run(self, inputs: Union[Dict, str, bytes]) -> Dict[str, Any]

Parameters:

Parameter	Type	Description
`inputs`	`Dict`, `str`, or `bytes`	Input data

Returns: Normalized response dictionary Response Format:

# Success
{
    "status": "success",
    "output": {...}  # Your result
}

# Error
{
    "status": "error",
    "error": "Error message"
}

Example:

result = deployed_program.run({"text": "hello"})

if result["status"] == "success":
    print(result["output"])
else:
    print(f"Error: {result['error']}")

stop()

Stop the software and cleanup resources.

@abstractmethod
def stop(self) -> None

Behavior by Strategy:

Strategy	What `stop()` Does
LOCAL	Unloads Python module from `sys.modules`
DOCKER	Stops and removes the Docker container
MODAL	Runs `modal app stop` to terminate the deployment
BENTOML	Runs `bentoml deployment terminate` to stop the service
LANGGRAPH	Deletes the conversation thread and disconnects

Example:

deployed_program.stop()
print(deployed_program.is_healthy())  # False

start()

Start or restart a stopped deployment.

@abstractmethod
def start(self) -> None

Behavior by Strategy:

Strategy	What `start()` Does
LOCAL	Reloads the Python module
DOCKER	Creates and starts a new container
MODAL	Re-lookups the Modal function
BENTOML	Re-deploys by running `deploy.py` from code_path
LANGGRAPH	Reconnects to LangGraph Platform

Example:

# After stop()
deployed_program.stop()
print(deployed_program.is_healthy())  # False

# Restart
deployed_program.start()
print(deployed_program.is_healthy())  # True

# Can run again
result = deployed_program.run({"text": "hello again"})

logs()

Get execution logs.

@abstractmethod
def logs(self) -> str

Returns: Log content as string Example:

print(deployed_program.logs())
# Initialized local deployment
# run() called with: {"text": "hello"}
# run() completed: status=success

is_healthy()

Check if software is running and healthy.

@abstractmethod
def is_healthy(self) -> bool

Returns: True if healthy, False otherwise Example:

if deployed_program.is_healthy():
    result = deployed_program.run(inputs)
else:
    print("Software not healthy!")

name (property)

Return the deployment strategy name.

@property
@abstractmethod
def name(self) -> str

Example:

print(deployed_program.name)  # "modal"

Convenience Methods

call()

Shorthand for run().

# These are equivalent
result = deployed_program.run(inputs)
result = deployed_program(inputs)

run_batch()

Run multiple inputs in sequence.

def run_batch(self, inputs_list: List[Any]) -> List[Dict[str, Any]]

Example:

results = deployed_program.run_batch([
    {"text": "hello"},
    {"text": "world"},
    {"text": "test"},
])

for result in results:
    print(result["output"])

Context Manager Support

with DeploymentFactory.create(strategy, config) as deployed_program:
    result = deployed_program.run(inputs)
# deployed_program.stop() called automatically

DeployedSoftware

Concrete implementation of Software. Wraps any runner with unified interface.

Additional Methods

get_adapted_path()

Get path to the adapted code.

def get_adapted_path(self) -> str

Example:

path = deployed_program.get_adapted_path()
print(path)  # "/path/to/solution_adapted_modal"

get_endpoint()

Get HTTP endpoint if applicable.

def get_endpoint(self) -> Optional[str]

Returns: Endpoint URL or None Example:

endpoint = deployed_program.get_endpoint()
if endpoint:
    print(f"API available at: {endpoint}")

get_deployment_info()

Get full deployment metadata.

def get_deployment_info(self) -> Dict[str, Any]

Returns:

{
    "strategy": "modal",
    "provider": "modal",
    "endpoint": "https://...",
    "adapted_path": "/path/to/...",
    "adapted_files": ["modal_app.py", "main.py"],
    "resources": {"gpu": "T4", "memory": "16Gi"},
}

get_strategy()

Get the deployment strategy name.

def get_strategy(self) -> str

get_provider()

Get the cloud provider name.

def get_provider(self) -> Optional[str]

Data Classes

DeploymentSetting

Result of strategy selection.

@dataclass
class DeploymentSetting:
    strategy: str              # "local", "docker", "modal", etc.
    provider: Optional[str]    # Cloud provider name
    resources: Dict[str, Any]  # Resource requirements
    interface: str             # "function", "http", etc.
    reasoning: str             # Why this was selected

AdaptationResult

Result of code adaptation.

@dataclass
class AdaptationResult:
    success: bool
    adapted_path: str           # Path to adapted repo
    run_interface: Dict[str, Any]
    files_changed: List[str] = field(default_factory=list)
    error: Optional[str] = None

DeploymentInfo

Metadata about the deployment.

@dataclass
class DeploymentInfo:
    strategy: str
    provider: Optional[str] = None
    endpoint: Optional[str] = None
    adapted_path: str = ""
    adapted_files: List[str] = field(default_factory=list)
    resources: Dict[str, Any] = field(default_factory=dict)

SelectorAgent

LLM-based strategy selection agent.

from kapso.deployment.selector.agent import SelectorAgent

Methods

select()

Select deployment configuration for a solution.

def select(
    self,
    solution: SolutionResult,
    allowed_strategies: Optional[List[str]] = None,
    resources: Optional[Dict[str, Any]] = None,
) -> DeploymentSetting

explain()

Get human-readable explanation of selection.

def explain(self, solution: SolutionResult) -> str

Example:

selector = SelectorAgent()
explanation = selector.explain(solution)
print(explanation)
# Deployment Selection Analysis
# ========================================
# 
# Strategy:  modal
# Interface: function
# Provider:  modal
# Resources: {'gpu': 'T4', 'memory': '16Gi'}
# 
# Reasoning: GPU workload detected (torch dependency)

AdapterAgent

Code transformation and deployment agent.

from kapso.deployment.adapter.agent import AdapterAgent

Methods

adapt()

Adapt a solution for deployment.

def adapt(
    self,
    solution: SolutionResult,
    setting: DeploymentSetting,
    allowed_strategies: Optional[List[str]] = None,
) -> AdaptationResult

Note: Creates a copy of the solution at {code_path}_adapted_{strategy}. Original is never modified.

StrategyRegistry

Auto-discovery system for deployment strategies.

from kapso.deployment.strategies import StrategyRegistry

Methods

get()

Get singleton registry instance.

@classmethod
def get(cls) -> StrategyRegistry

list_strategies()

List available strategy names.

def list_strategies(self, allowed: Optional[List[str]] = None) -> List[str]

get_strategy()

Get strategy config by name.

def get_strategy(self, name: str) -> DeployStrategyConfig

get_selector_instruction()

Get selector instruction content.

def get_selector_instruction(self, name: str) -> str

get_adapter_instruction()

Get adapter instruction content.

def get_adapter_instruction(self, name: str) -> str

get_runner_class()

Get the runner class for a strategy.

def get_runner_class(self, name: str) -> type

get_default_run_interface()

Get default run interface from config.yaml.

def get_default_run_interface(self, name: str) -> Dict[str, Any]

Runner (Abstract Base Class)

Base class for strategy-specific runners.

from kapso.deployment.strategies.base import Runner

Lifecycle

Runners follow this lifecycle:

__init__() → Ready state
run() → Execute (can be called multiple times)
stop() → Stopped state (resources cleaned up)
start() → Ready state (can run again)

┌─────────────┐    stop()    ┌─────────────┐
│   READY     │ ──────────>  │   STOPPED   │
│ is_healthy  │              │ !is_healthy │
│   = True    │  <──────────  │   = False   │
└─────────────┘    start()   └─────────────┘

Abstract Methods

class Runner(ABC):
    @abstractmethod
    def run(self, inputs: Union[Dict, str, bytes]) -> Any:
        """Execute with inputs and return result."""
        pass
    
    @abstractmethod
    def start(self) -> None:
        """Start or restart the runner."""
        pass
    
    @abstractmethod
    def stop(self) -> None:
        """Stop and cleanup resources."""
        pass
    
    @abstractmethod
    def is_healthy(self) -> bool:
        """Check if runner is healthy."""
        pass
    
    def get_logs(self) -> str:
        """Get runner logs (default: empty)."""
        return ""

Complete Usage Example

from kapso.kapso import Kapso
from kapso.deployment import DeploymentFactory, DeployStrategy, DeployConfig

# 1. Build a solution
kapso = Kapso()
solution = kapso.evolve("Create a text embedding API using sentence-transformers")

# 2. Deploy with auto-selection
deployed_program = kapso.deploy(solution)
print(f"Deployed using: {deployed_program.name}")
print(f"Endpoint: {deployed_program.get_endpoint()}")

# 3. Use the deployed_program
result = deployed_program.run({"text": "Hello world"})
if result["status"] == "success":
    embeddings = result["output"]["embeddings"]
    print(f"Got {len(embeddings)} dimensional embeddings")

# 4. Batch processing
texts = ["text 1", "text 2", "text 3"]
results = deployed_program.run_batch([{"text": t} for t in texts])

# 5. Check health
if deployed_program.is_healthy():
    print("Service is healthy")

# 6. Get logs for debugging
print(deployed_program.logs())

# 7. Get deployment details
info = deployed_program.get_deployment_info()
print(f"Strategy: {info['strategy']}")
print(f"Adapted path: {info['adapted_path']}")

# 8. Stop and restart (full lifecycle)
deployed_program.stop()
print(f"Is healthy after stop: {deployed_program.is_healthy()}")  # False

deployed_program.start()
print(f"Is healthy after start: {deployed_program.is_healthy()}")  # True

# 9. Run again after restart
result = deployed_program.run({"text": "Working again!"})

# 10. Final cleanup
deployed_program.stop()

Error Handling

try:
    deployed_program = DeploymentFactory.create(DeployStrategy.MODAL, config)
    result = deployed_program.run(inputs)
    
    if result["status"] == "error":
        print(f"Execution error: {result['error']}")
        
except ValueError as e:
    print(f"Invalid configuration: {e}")
    
except RuntimeError as e:
    print(f"Deployment failed: {e}")
    
finally:
    if 'deployed_program' in locals():
        deployed_program.stop()

Environment Variables

Variable	Strategy	Description
`MODAL_TOKEN_ID`	Modal	Modal authentication
`MODAL_TOKEN_SECRET`	Modal	Modal authentication
`LANGSMITH_API_KEY`	LangGraph	LangSmith API key
`BENTOML_API_TOKEN`	BentoML	BentoCloud authentication

​Module Exports

​DeploymentFactory

​Methods

​create()

​list_strategies()

​explain_strategy()

​print_strategies_info()

​DeployStrategy

​DeployConfig

​Software (Abstract Base Class)

​Methods

​run()

​stop()

​start()

​logs()

​is_healthy()

​name (property)

​Convenience Methods

​call()

​run_batch()

​Context Manager Support

​DeployedSoftware

​Additional Methods

​get_adapted_path()

​get_endpoint()

​get_deployment_info()

​get_strategy()

​get_provider()

​Data Classes

​DeploymentSetting

​AdaptationResult

​DeploymentInfo

​SelectorAgent

​Methods

​select()

​explain()

​AdapterAgent

​Methods

​adapt()

​StrategyRegistry

​Methods

​get()

​list_strategies()

​get_strategy()

​get_selector_instruction()

​get_adapter_instruction()

​get_runner_class()

​get_default_run_interface()

​Runner (Abstract Base Class)

​Lifecycle

​Abstract Methods

​Complete Usage Example

​Error Handling

​Environment Variables

Module Exports

DeploymentFactory

Methods

create()

list_strategies()

explain_strategy()

print_strategies_info()

DeployStrategy

DeployConfig

Software (Abstract Base Class)

Methods

run()

stop()

start()

logs()

is_healthy()

name (property)

Convenience Methods

call()

run_batch()

Context Manager Support

DeployedSoftware

Additional Methods

get_adapted_path()

get_endpoint()

get_deployment_info()

get_strategy()

get_provider()

Data Classes

DeploymentSetting

AdaptationResult

DeploymentInfo

SelectorAgent

Methods

select()

explain()

AdapterAgent

Methods

adapt()

StrategyRegistry

Methods

get()

list_strategies()

get_strategy()

get_selector_instruction()

get_adapter_instruction()

get_runner_class()

get_default_run_interface()

Runner (Abstract Base Class)

Lifecycle

Abstract Methods

Complete Usage Example

Error Handling

Environment Variables