Deploy K8s¶

Deploy K8s — mvp.deploy_k8s

Cluster: Core Infrastructure | Type: component | MCP Tools: None

Overview¶

Kubernetes deployment provider that generates YAML manifests (Deployment, Service, ConfigMap, Namespace) and applies them via kubectl. Supports configurable replicas, resource requests/limits, service types, K8s secret injection, and rollout policies with automatic rollback on failure. The K8sOrchestrator implements the shared Orchestrator protocol so it composes naturally with other deployment components.

Launch-plan scope

deploy_k8s is an engineer-facing building block for teams that already operate Kubernetes. It is not the recommended first-user or Phase 0 launch deployment path. The current launch plan favors a simple VPS/Docker deployment for early users and explicitly treats Kubernetes as overkill before meaningful scale. Use this component for existing clusters, GitOps workflows, EKS/GKE handoff, or later production hardening, not for the 10-minute non-developer onboarding path.

When to use:

Deploying a G6 base to an existing Kubernetes cluster with production-grade manifests
Generating and committing K8s YAML as a GitOps artefact from a CI/CD pipeline
Rolling back a failed deployment or inspecting pod status from an agent workflow

Do not use as the default launch path when:

The target is a first pilot, clean-machine demo, or non-developer onboarding flow
A single VPS with Docker Compose is enough for web, MCP, REST, Postgres, and Redis
The operator does not already have cluster ownership, RBAC, ingress, DNS, TLS, secrets, backup, and monitoring practices in place

Example:

from mvp.deploy_k8s import K8sBlock, K8sInput, K8sManifestConfig
from mvp.deploy_core.schema import ImageSpec, DeployTarget

block = K8sBlock(name="k8s")
result = block.infer(K8sInput(
    op="generate",
    image=ImageSpec(name="g6-rest", tag="abc1234"),
    target=DeployTarget(provider="k8s", namespace="g6-prod"),
    config=K8sManifestConfig(replicas=3, service_type="LoadBalancer"),
))
# result.ok → True; result.value.artifact → path to generated YAML

Works well with: deploy_core, deploy_docker, cicd

Public API¶

`DeployK8sDecisionError(ValueError)`¶

The LLM did not produce a usable, validated deploy_k8s decision.

`ApplyReadinessDecision`¶

Field	Type	Default
`approved`	`bool`	required
`blocking_reasons`	`list[str]`	field(default_factory=list)
`warnings`	`list[str]`	field(default_factory=list)
`rationale`	`str`	''
`confidence`	`float`	0.0
`completion_state`	`str`	'qualified-draft'
`degraded`	`bool`	False
`raw_response`	`str`	''

`DeployK8sRuntime(Protocol)`¶

Methods:

`assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], deterministic: ApplyReadinessDecision) -> ApplyReadinessDecision`¶

`LLMDeployK8sRuntime`¶

Provider-neutral deploy_k8s runtime backed by G6's LLM caller interface.

Constructor:

Parameter	Type	Default
`llm`	`LLMCaller \\| None`	None

Methods:

`assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], deterministic: ApplyReadinessDecision) -> ApplyReadinessDecision`¶

`DeployK8sPlanner`¶

Runtime-first facade with a real deterministic fallback and safety floor.

Constructor:

Parameter	Type	Default
`runtime`	`DeployK8sRuntime \\| None`	None

Methods:

`assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], check_rbac: bool = True) -> ApplyReadinessDecision`¶

`K8sBlock(AIBlock[K8sInput, DeployResult, None])`¶

Field	Type	Default
`name`	`str`	'k8s'

Methods:

`infer(data: K8sInput) -> Result[DeployResult]`¶

`K8sOrchestrator`¶

Orchestrator implementation using kubectl CLI.

Constructor:

Parameter	Type	Default
`kubeconfig`	`str \\| None`	None
`dry_run`	`bool`	False

Methods:

`generate_manifests(image: ImageSpec, target: DeployTarget, config: K8sManifestConfig | None = None) -> DeployResult`¶

`apply(manifest_path: str, image: ImageSpec | None = None, target: DeployTarget | None = None) -> DeployResult`¶

`status(namespace: str) -> DeployResult`¶

`wait_healthy(image: ImageSpec, target: DeployTarget) -> DeployResult`¶

Poll kubectl rollout status until deployment is healthy or timeout.

`rollback(image: ImageSpec | str, target: DeployTarget | str) -> DeployResult`¶

kubectl rollout undo.

`DeployK8sPatternRuntime`¶

Load-bearing context-minimization mechanism for the deploy_k8s runtime.

Field	Type	Default
`manifest_document_limit`	`int`	32

Methods:

`minimize_apply_context(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]]) -> dict[str, Any]`¶

Send only the decision-relevant apply fields to the LLM.

`K8sPolicyConstraints`¶

Caller-supplied policy constraints checked before cluster mutation.

Field	Type	Default
`allowed_namespaces`	`tuple[str, ...]`	()
`allowed_image_registries`	`tuple[str, ...]`	()
`allowed_rollout_failure_actions`	`tuple[str, ...]`	()
`max_replicas`	`int \\| None`	None
`required_service_account`	`str`	''
`production`	`bool`	False
`production_approved`	`bool`	False

`K8sManifestConfig`¶

Configuration for K8s manifest generation.

Field	Type	Default
`namespace`	`str`	'g6'
`replicas`	`int`	1
`container_port`	`int`	8000
`service_type`	`str`	'ClusterIP'
`resources`	`dict`	field(default_factory=lambda: {'requests': {'cpu': '100m', 'memory': '128Mi'}, 'limits': {'cpu': '500m', 'memory': '512Mi'}})
`env`	`dict[str, str]`	field(default_factory=dict)
`output_dir`	`str`	'k8s/'
`secrets`	`list[SecretRef]`	field(default_factory=list)
`rollout_policy`	`RolloutPolicy`	field(default_factory=RolloutPolicy)

`K8sInput`¶

Input for K8sBlock.infer().

Field	Type	Default
`op`	`str`	required
`image`	`ImageSpec`	field(default_factory=lambda: ImageSpec(name=''))
`target`	`DeployTarget`	field(default_factory=lambda: DeployTarget(provider='k8s'))
`manifest_path`	`str`	''
`deployment`	`str`	''
`config`	`K8sManifestConfig`	field(default_factory=K8sManifestConfig)
`policy_constraints`	`K8sPolicyConstraints`	field(default_factory=K8sPolicyConstraints)

Functions¶

`agentic_planner_enabled(default_enabled: bool = True) -> bool`¶

Decide whether the agentic deploy_k8s planner should be used.

`validate_apply_readiness(decision: ApplyReadinessDecision) -> None`¶

Fail-closed validation of an apply-readiness decision (mirrors deploy_core).

`deterministic_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]] | None = None, check_rbac: bool = True, policy_constraints: K8sPolicyConstraints | None = None) -> ApplyReadinessDecision`¶

The SINGLE SOURCE OF TRUTH apply-readiness safety floor.

`generate_namespace(namespace: str) -> dict`¶

`generate_deployment(image: ImageSpec, config: K8sManifestConfig) -> dict`¶

`generate_service(image: ImageSpec, config: K8sManifestConfig) -> dict | None`¶

`generate_configmap(image: ImageSpec, env: dict[str, str], namespace: str = 'default') -> dict`¶

`generate_all(image: ImageSpec, config: K8sManifestConfig) -> str`¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for the deploy_k8s-applied patterns.

`get_skill_catalog() -> DeployK8sSkillCatalog`¶

`list_patterns() -> dict[str, Any]`¶

Return the deploy_k8s applied-pattern + skill surface (block list_patterns op).

Deploy K8s¶

Overview¶

Public API¶

DeployK8sDecisionError(ValueError)¶

ApplyReadinessDecision¶

DeployK8sRuntime(Protocol)¶

assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], deterministic: ApplyReadinessDecision) -> ApplyReadinessDecision¶

LLMDeployK8sRuntime¶

assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], deterministic: ApplyReadinessDecision) -> ApplyReadinessDecision¶

DeployK8sPlanner¶

assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], check_rbac: bool = True) -> ApplyReadinessDecision¶

K8sBlock(AIBlock[K8sInput, DeployResult, None])¶

infer(data: K8sInput) -> Result[DeployResult]¶

K8sOrchestrator¶

generate_manifests(image: ImageSpec, target: DeployTarget, config: K8sManifestConfig | None = None) -> DeployResult¶

apply(manifest_path: str, image: ImageSpec | None = None, target: DeployTarget | None = None) -> DeployResult¶

status(namespace: str) -> DeployResult¶

wait_healthy(image: ImageSpec, target: DeployTarget) -> DeployResult¶

rollback(image: ImageSpec | str, target: DeployTarget | str) -> DeployResult¶

DeployK8sPatternRuntime¶

minimize_apply_context(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]]) -> dict[str, Any]¶

K8sPolicyConstraints¶

K8sManifestConfig¶

K8sInput¶

Functions¶

agentic_planner_enabled(default_enabled: bool = True) -> bool¶

validate_apply_readiness(decision: ApplyReadinessDecision) -> None¶

deterministic_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]] | None = None, check_rbac: bool = True, policy_constraints: K8sPolicyConstraints | None = None) -> ApplyReadinessDecision¶

generate_namespace(namespace: str) -> dict¶

generate_deployment(image: ImageSpec, config: K8sManifestConfig) -> dict¶

generate_service(image: ImageSpec, config: K8sManifestConfig) -> dict | None¶

generate_configmap(image: ImageSpec, env: dict[str, str], namespace: str = 'default') -> dict¶

generate_all(image: ImageSpec, config: K8sManifestConfig) -> str¶

applied_agentic_patterns() -> list[dict[str, Any]]¶

get_skill_catalog() -> DeployK8sSkillCatalog¶

list_patterns() -> dict[str, Any]¶

`DeployK8sDecisionError(ValueError)`¶

`ApplyReadinessDecision`¶

`DeployK8sRuntime(Protocol)`¶

`assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], deterministic: ApplyReadinessDecision) -> ApplyReadinessDecision`¶

`LLMDeployK8sRuntime`¶

`assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], deterministic: ApplyReadinessDecision) -> ApplyReadinessDecision`¶

`DeployK8sPlanner`¶

`assess_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]], check_rbac: bool = True) -> ApplyReadinessDecision`¶

`K8sBlock(AIBlock[K8sInput, DeployResult, None])`¶

`infer(data: K8sInput) -> Result[DeployResult]`¶

`K8sOrchestrator`¶

`generate_manifests(image: ImageSpec, target: DeployTarget, config: K8sManifestConfig | None = None) -> DeployResult`¶

`apply(manifest_path: str, image: ImageSpec | None = None, target: DeployTarget | None = None) -> DeployResult`¶

`status(namespace: str) -> DeployResult`¶

`wait_healthy(image: ImageSpec, target: DeployTarget) -> DeployResult`¶

`rollback(image: ImageSpec | str, target: DeployTarget | str) -> DeployResult`¶

`DeployK8sPatternRuntime`¶

`minimize_apply_context(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]]) -> dict[str, Any]`¶

`K8sPolicyConstraints`¶

`K8sManifestConfig`¶

`K8sInput`¶

`agentic_planner_enabled(default_enabled: bool = True) -> bool`¶

`validate_apply_readiness(decision: ApplyReadinessDecision) -> None`¶

`deterministic_apply_readiness(namespace: str, service_account: str, manifest_documents: list[dict[str, Any]] | None = None, check_rbac: bool = True, policy_constraints: K8sPolicyConstraints | None = None) -> ApplyReadinessDecision`¶

`generate_namespace(namespace: str) -> dict`¶

`generate_deployment(image: ImageSpec, config: K8sManifestConfig) -> dict`¶

`generate_service(image: ImageSpec, config: K8sManifestConfig) -> dict | None`¶

`generate_configmap(image: ImageSpec, env: dict[str, str], namespace: str = 'default') -> dict`¶

`generate_all(image: ImageSpec, config: K8sManifestConfig) -> str`¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

`get_skill_catalog() -> DeployK8sSkillCatalog`¶

`list_patterns() -> dict[str, Any]`¶