Skip to main content

Module Exports

from src.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:
ParameterTypeDescription
strategyDeployStrategyDeployment target (AUTO, LOCAL, DOCKER, etc.)
configDeployConfigDeployment configuration
strategiesList[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 src.deployment import DeploymentFactory, DeployStrategy, DeployConfig

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

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

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

# Restrict to certain strategies
software = 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:
ParameterTypeDescription
strategystrStrategy 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 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 src.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:
FieldTypeDefaultDescription
solutionSolutionResultRequiredThe built solution from Tinkerer.evolve()
env_varsDict[str, str]{}Environment variables
timeoutint300Execution timeout in seconds
coding_agentstr"claude_code"Coding agent for adaptation
Properties:
PropertyTypeDescription
code_pathstrPath to the generated code
goalstrThe original goal/objective
Example:
from src.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:
ParameterTypeDescription
inputsDict, str, or bytesInput data
Returns: Normalized response dictionary Response Format:
# Success
{
    "status": "success",
    "output": {...}  # Your result
}

# Error
{
    "status": "error",
    "error": "Error message"
}
Example:
result = software.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:
StrategyWhat stop() Does
LOCALUnloads Python module from sys.modules
DOCKERStops and removes the Docker container
MODALRuns modal app stop to terminate the deployment
BENTOMLRuns bentoml deployment terminate to stop the service
LANGGRAPHDeletes the conversation thread and disconnects
Example:
software.stop()
print(software.is_healthy())  # False

start()

Start or restart a stopped deployment.
@abstractmethod
def start(self) -> None
Behavior by Strategy:
StrategyWhat start() Does
LOCALReloads the Python module
DOCKERCreates and starts a new container
MODALRe-lookups the Modal function
BENTOMLRe-deploys by running deploy.py from code_path
LANGGRAPHReconnects to LangGraph Platform
Example:
# After stop()
software.stop()
print(software.is_healthy())  # False

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

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

logs()

Get execution logs.
@abstractmethod
def logs(self) -> str
Returns: Log content as string Example:
print(software.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 software.is_healthy():
    result = software.run(inputs)
else:
    print("Software not healthy!")

name (property)

Return the deployment strategy name.
@property
@abstractmethod
def name(self) -> str
Example:
print(software.name)  # "modal"

Convenience Methods

call()

Shorthand for run().
# These are equivalent
result = software.run(inputs)
result = software(inputs)

run_batch()

Run multiple inputs in sequence.
def run_batch(self, inputs_list: List[Any]) -> List[Dict[str, Any]]
Example:
results = software.run_batch([
    {"text": "hello"},
    {"text": "world"},
    {"text": "test"},
])

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

Context Manager Support

with DeploymentFactory.create(strategy, config) as software:
    result = software.run(inputs)
# software.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 = software.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 = software.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 src.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 src.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 src.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 src.deployment.strategies.base import Runner

Lifecycle

Runners follow this lifecycle:
  1. __init__() → Ready state
  2. run() → Execute (can be called multiple times)
  3. stop() → Stopped state (resources cleaned up)
  4. 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 src.tinkerer import Tinkerer
from src.deployment import DeploymentFactory, DeployStrategy, DeployConfig

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

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

# 3. Use the software
result = software.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 = software.run_batch([{"text": t} for t in texts])

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

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

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

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

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

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

# 10. Final cleanup
software.stop()

Error Handling

try:
    software = DeploymentFactory.create(DeployStrategy.MODAL, config)
    result = software.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 'software' in locals():
        software.stop()

Environment Variables

VariableStrategyDescription
MODAL_TOKEN_IDModalModal authentication
MODAL_TOKEN_SECRETModalModal authentication
LANGSMITH_API_KEYLangGraphLangSmith API key
BENTOML_API_TOKENBentoMLBentoCloud authentication