Skip to main content

Overview

After evolve() generates a solution, deploy() turns it into running software. The deployment pipeline handles adaptation, strategy selection, and execution.

Deployment Flow

  1. Selector: Analyzes code and chooses deployment strategy
  2. Adapter: Modifies code for the chosen strategy
  3. Runner: Executes the adapted code
  4. Software: Unified interface for running and managing

Quick Start

from kapso.kapso import Kapso, DeployStrategy

kapso = Kapso()
solution = kapso.evolve(goal="Build a classifier")

# Deploy with explicit strategy
deployed_program = kapso.deploy(solution, strategy=DeployStrategy.LOCAL)

# Run inference
result = deployed_program.run({"data_path": "./test.csv"})
print(result)  # {"status": "success", "output": ...}

# Lifecycle management
deployed_program.stop()
deployed_program.start()
deployed_program.stop()

Deployment Strategies

StrategyDescriptionBest For
AUTOSystem chooses best strategyDefault
LOCALRun as local Python processDevelopment, testing
DOCKERRun in Docker containerIsolation, reproducibility
MODALDeploy to Modal.comServerless, GPU
BENTOMLDeploy with BentoMLProduction ML
LANGGRAPHDeploy as LangGraph agentAgent workflows

DeployStrategy Enum

from kapso.deployment import DeployStrategy

DeployStrategy.AUTO      # System selects
DeployStrategy.LOCAL     # Local process
DeployStrategy.DOCKER    # Docker container
DeployStrategy.MODAL     # Modal.com
DeployStrategy.BENTOML   # BentoML
DeployStrategy.LANGGRAPH # LangGraph

Software Interface

All deployments return a Software instance with a unified interface:
class Software:
    def run(self, inputs: Dict) -> Dict:
        """Execute with input data."""
        # Returns: {"status": "success", "output": ...}
        # Or: {"status": "error", "error": ...}

    def stop(self) -> None:
        """Stop and cleanup resources."""

    def start(self) -> None:
        """Restart a stopped deployment."""

    def is_healthy(self) -> bool:
        """Check if service is running."""

    def logs(self) -> str:
        """Get execution logs."""

Response Format

All run() calls return a consistent format:
# Success
{"status": "success", "output": <result>}

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

Configuration

from kapso.deployment import DeployConfig

config = DeployConfig(
    solution=solution,
    env_vars={"API_KEY": "..."},
    coding_agent="claude_code",
)

deployed_program = DeploymentFactory.create(DeployStrategy.LOCAL, config)

Deployment Info

Get details about the deployment:
# Get deployment metadata
info = deployed_program.get_deployment_info()
print(info)
# {
#     "strategy": "local",
#     "provider": None,
#     "endpoint": None,
#     "adapted_path": "/path/to/adapted",
#     "adapted_files": ["main.py", "run.py"],
#     "resources": {}
# }

# Individual accessors
deployed_program.get_strategy()      # "local"
deployed_program.get_adapted_path()  # Path to adapted code
deployed_program.get_endpoint()      # HTTP endpoint (if applicable)

Lifecycle Management

deployed_program = kapso.deploy(solution, strategy=DeployStrategy.LOCAL)

# Initial run
result = deployed_program.run({"data": "..."})

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

# Stop (cleanup resources)
deployed_program.stop()

# Restart
deployed_program.start()

# Run again
result = deployed_program.run({"data": "..."})

# Final cleanup
deployed_program.stop()

Error Handling

result = deployed_program.run({"data": "..."})

if result["status"] == "error":
    print(f"Error: {result['error']}")
    print(f"Logs: {deployed_program.logs()}")

Next Steps