Add a custom deployment strategy to Kapso, step by step

src/kapso/deployment/strategies/my_strategy/
├── __init__.py
├── runner.py              # Runner implementation
├── config.yaml            # Strategy configuration
├── selector_instruction.txt   # Instructions for auto-selection
└── adapter_instruction.txt    # Instructions for code adaptation

mkdir -p src/kapso/deployment/strategies/my_strategy

from kapso.deployment.strategies.base import Runner
from typing import Any, Dict

class MyStrategyRunner(Runner):
    """Runner for my custom strategy."""

    def __init__(
        self,
        code_path: str,
        module: str = "main",
        callable: str = "predict",
        **kwargs,
    ):
        self.code_path = code_path
        self.module = module
        self.callable = callable
        self._running = False

    def run(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """Execute with inputs."""
        if not self._running:
            self.start()

        try:
            # Your execution logic here
            result = self._execute(inputs)
            return {"status": "success", "output": result}
        except Exception as e:
            return {"status": "error", "error": str(e)}

    def start(self) -> None:
        """Start the service."""
        # Initialize resources
        self._running = True

    def stop(self) -> None:
        """Stop and cleanup."""
        # Cleanup resources
        self._running = False

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

    def get_logs(self) -> str:
        """Return execution logs."""
        return ""

    def _execute(self, inputs: Dict) -> Any:
        """Internal execution logic."""
        # Implement your execution here
        pass

name: my_strategy
provider: my_provider
interface: function

# Runner class name (in runner.py) — required; the registry
# refuses a strategy without it
runner_class: MyStrategyRunner

# Default resource requirements
default_resources:
  cpu: "1"
  memory: "1Gi"

# Default run interface (passed to the runner constructor)
run_interface:
  type: "function"
  module: "main"
  callable: "predict"

# The tool preflight and `kapso doctor deploy` check for (omit for in-process targets)
requires:
  binary: mycloud
  install: "pip install mycloud && mycloud login"

# Files your adapter instruction asks the agent to create; the validator fails
# the adaptation without them, and the agent is asked to fix it
required_files: [main.py, mycloud.yaml]

## My Strategy

### Summary

A custom deployment strategy for [specific use case].

### When to Use

- Condition 1
- Condition 2
- Condition 3

### When NOT to Use

- Anti-condition 1
- Anti-condition 2

### Requirements

- Requirement 1
- Requirement 2

## Adaptation Instructions for My Strategy

### Required Files

1. `run.py` - Entry point script
2. `requirements.txt` - Dependencies

### Entry Point Format

```python
def predict(inputs: dict) -> dict:
    """Main prediction function."""
    # Implementation
    return {"result": ...}
```

### Environment Variables

- `MY_STRATEGY_CONFIG` - Configuration path

### Notes

- Keep dependencies minimal
- Handle errors gracefully

from .runner import MyStrategyRunner

__all__ = ["MyStrategyRunner"]

from abc import ABC, abstractmethod
from typing import Any, Dict, Union

class Runner(ABC):
    """Base class for deployment runners."""

    @abstractmethod
    def run(self, inputs: Union[Dict, str, bytes]) -> Any:
        """Execute with inputs; DeployedSoftware normalizes the result."""
        pass

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

    @abstractmethod
    def start(self) -> None:
        """Start or restart the service."""
        pass

    @abstractmethod
    def is_healthy(self) -> bool:
        """Check if service is running and ready."""
        pass

    def get_logs(self) -> str:
        """Return execution logs."""
        return ""

from kapso.deployment.strategies import StrategyRegistry

registry = StrategyRegistry.get()

# List all strategies
strategies = registry.list_strategies()
print(strategies)  # ["bentoml", "docker", "langgraph", "local", "modal", "my_strategy"]

# Check if strategy exists
exists = registry.strategy_exists("my_strategy")

# Get runner class
runner_class = registry.get_runner_class("my_strategy")

# Get instructions
selector_inst = registry.get_selector_instruction("my_strategy")
adapter_inst = registry.get_adapter_instruction("my_strategy")

from kapso import Kapso
from kapso.deployment import DeployStrategy

# The enum member appears automatically: DeployStrategy rebuilds
# itself from every strategies/ directory that has a config.yaml
deployed_program = kapso.deploy(solution, strategy=DeployStrategy.MY_STRATEGY)

# Or via factory
from kapso.deployment import DeploymentFactory, DeployConfig

config = DeployConfig(solution=solution)
deployed_program = DeploymentFactory.create(DeployStrategy.MY_STRATEGY, config)

import pytest
from kapso.deployment.strategies.my_strategy.runner import MyStrategyRunner

def test_runner_lifecycle():
    runner = MyStrategyRunner(
        code_path="/path/to/code",
        module="main",
        callable="predict",
    )

    # Test start
    runner.start()
    assert runner.is_healthy()

    # Test run
    result = runner.run({"input": "test"})
    assert result["status"] == "success"

    # Test stop
    runner.stop()
    assert not runner.is_healthy()

import subprocess
import requests
from kapso.deployment.strategies.base import Runner

class HttpRunner(Runner):
    def __init__(self, code_path: str, port: int = 8000, **kwargs):
        self.code_path = code_path
        self.port = port
        self.process = None
        self.endpoint = f"http://localhost:{port}"

    def start(self) -> None:
        self.process = subprocess.Popen(
            ["python", "-m", "uvicorn", "main:app", "--port", str(self.port)],
            cwd=self.code_path,
        )
        # Wait for server to be ready
        self._wait_for_ready()

    def run(self, inputs: dict) -> dict:
        response = requests.post(f"{self.endpoint}/predict", json=inputs)
        return response.json()

    def stop(self) -> None:
        if self.process:
            self.process.terminate()
            self.process.wait()
            self.process = None

    def is_healthy(self) -> bool:
        try:
            response = requests.get(f"{self.endpoint}/health")
            return response.status_code == 200
        except:
            return False