diff --git a/.gitignore b/.gitignore index bb9e3a3..4e5ff1d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ /target pipelit-client/target .sisyphus/ + +.env +.omc/ diff --git a/Dockerfile b/Dockerfile index 1d37d99..eca5947 100644 --- a/Dockerfile +++ b/Dockerfile @@ -46,10 +46,21 @@ RUN ARCH=$(uname -m) && \ mv "/root/.config/plit/dragonfly-${ARCH}" /root/.config/plit/dragonfly && \ chmod +x /root/.config/plit/dragonfly +# Download agentgateway binary +RUN ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/') && \ + curl -fSL "https://github.com/agentgateway/agentgateway/releases/download/v1.0.1/agentgateway-linux-${ARCH}" \ + -o /usr/local/bin/agentgateway && \ + chmod +x /usr/local/bin/agentgateway + +# Install yq for config assembly +RUN curl -fSL "https://github.com/mikefarah/yq/releases/latest/download/yq_linux_$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/')" \ + -o /usr/local/bin/yq && \ + chmod +x /usr/local/bin/yq + COPY docker/entrypoint.sh /entrypoint.sh RUN chmod +x /entrypoint.sh -EXPOSE 8080 8000 +EXPOSE 8080 8000 4000 3000 15000 VOLUME ["/root/.local/share/plit", "/root/.config/pipelit/workspaces"] diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 2f23669..6c927f1 100644 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -23,5 +23,84 @@ if [ ! -f "$CONFIG_FILE" ]; then eval plit init $INIT_ARGS fi +# Start agentgateway if configured +# Check both plit .env and pipelit .env for AGENTGATEWAY_DIR +AGW_DIR="" +PLIT_ENV="/root/.config/plit/.env" +PIPELIT_ENV="/root/.local/share/plit/pipelit/.env" +for envfile in "$PLIT_ENV" "$PIPELIT_ENV"; do + if [ -f "$envfile" ] && [ -z "$AGW_DIR" ]; then + AGW_DIR=$(grep "^AGENTGATEWAY_DIR=" "$envfile" | cut -d= -f2 | tr -d '"') + fi +done + +# Copy agentgateway env vars to Pipelit's .env if not already there +if [ -n "$AGW_DIR" ] && [ -f "$PIPELIT_ENV" ]; then + for var in AGENTGATEWAY_ENABLED AGENTGATEWAY_URL AGENTGATEWAY_DIR JWT_PRIVATE_KEY; do + if grep -q "^${var}=" "$PLIT_ENV" 2>/dev/null && ! grep -q "^${var}=" "$PIPELIT_ENV" 2>/dev/null; then + grep "^${var}=" "$PLIT_ENV" >> "$PIPELIT_ENV" + fi + done + # JWT_PRIVATE_KEY may be multiline — handle separately + if grep -q "^JWT_PRIVATE_KEY=" "$PLIT_ENV" 2>/dev/null && ! grep -q "^JWT_PRIVATE_KEY=" "$PIPELIT_ENV" 2>/dev/null; then + python3 -c " +import re +with open('$PLIT_ENV') as f: content = f.read() +m = re.search(r'(JWT_PRIVATE_KEY=\".*?\")', content, re.DOTALL) +if m: + with open('$PIPELIT_ENV', 'a') as f: f.write('\n' + m.group(1) + '\n') +" + fi +fi + +if [ -n "$AGW_DIR" ] && [ -d "$AGW_DIR" ]; then + echo "Starting agentgateway..." + + export AGENTGATEWAY_DIR="$AGW_DIR" + export AGENTGATEWAY_URL="${AGENTGATEWAY_URL:-http://localhost:4000}" + + # Ensure binary is in place + if [ ! -f "$AGW_DIR/bin/agentgateway" ]; then + mkdir -p "$AGW_DIR/bin" + cp /usr/local/bin/agentgateway "$AGW_DIR/bin/agentgateway" + fi + + # Get encryption key for key decryption (check both .env files) + FIELD_ENC_KEY="" + for envfile in "$PIPELIT_ENV" "$PLIT_ENV"; do + if [ -f "$envfile" ] && [ -z "$FIELD_ENC_KEY" ]; then + FIELD_ENC_KEY=$(grep "^FIELD_ENCRYPTION_KEY=" "$envfile" | head -1 | cut -d= -f2 | tr -d '"') + fi + done + + # Start agentgateway in background + ( + cd "$AGW_DIR" && \ + FIELD_ENCRYPTION_KEY="$FIELD_ENC_KEY" \ + PYTHON=/root/.local/share/plit/venv/bin/python3 \ + YQ=yq \ + ./start.sh > /tmp/agw.log 2>&1 + ) & + + # Block until agentgateway is ready. agentgateway is a hard boot + # dependency — if it never comes up, plit must not start serving. + AGW_READY=false + AGW_READY_TIMEOUT=45 + for i in $(seq 1 "$AGW_READY_TIMEOUT"); do + if curl -s -o /dev/null "$AGENTGATEWAY_URL/" 2>/dev/null; then + echo "agentgateway ready" + AGW_READY=true + break + fi + sleep 1 + done + + if [ "$AGW_READY" != true ]; then + echo "FATAL: agentgateway did not become ready within ${AGW_READY_TIMEOUT}s (${AGENTGATEWAY_URL}) — refusing to start plit without it" >&2 + cat /tmp/agw.log >&2 2>/dev/null || true + exit 1 + fi +fi + echo "Starting plit stack..." exec plit start --foreground diff --git a/src/commands/api.rs b/src/commands/api.rs new file mode 100644 index 0000000..edb531c --- /dev/null +++ b/src/commands/api.rs @@ -0,0 +1,194 @@ +use anyhow::{Context, Result}; +use clap::Subcommand; +use pipelit_client::apis::{configuration::Configuration, workflows_api}; +use pipelit_client::models::ValidateDslIn; + +use super::auth::pipelit_config; + +#[derive(Subcommand)] +pub enum ApiCommands { + #[command(subcommand)] + Workflow(WorkflowCommands), + + NodeTypes, +} + +#[derive(Subcommand)] +pub enum WorkflowCommands { + List { + #[arg(long, default_value = "50")] + limit: i32, + }, + + Get { + slug: String, + }, + + Validate { + #[arg(long)] + yaml: String, + }, + + Delete { + slug: String, + + #[arg(long)] + force: bool, + }, +} + +pub async fn run(cmd: ApiCommands, json_output: bool) -> Result<()> { + let config = pipelit_config()?; + + match cmd { + ApiCommands::Workflow(wf_cmd) => run_workflow(wf_cmd, &config, json_output).await, + ApiCommands::NodeTypes => run_node_types(&config, json_output).await, + } +} + +async fn run_workflow( + cmd: WorkflowCommands, + config: &Configuration, + json_output: bool, +) -> Result<()> { + match cmd { + WorkflowCommands::List { limit } => { + let result = + workflows_api::list_workflows_api_v1_workflows_get(config, Some(limit), Some(0)) + .await + .context("Failed to list workflows")?; + + if json_output { + println!("{}", serde_json::to_string_pretty(&result)?); + } else { + if let Some(arr) = result.as_array() { + println!("{:<8} {:<30} {:<20}", "ID", "NAME", "SLUG"); + println!("{}", "-".repeat(60)); + for wf in arr { + println!( + "{:<8} {:<30} {:<20}", + wf["id"].as_i64().unwrap_or(0), + truncate(wf["name"].as_str().unwrap_or("-"), 28), + wf["slug"].as_str().unwrap_or("-"), + ); + } + println!("\nTotal: {} workflows", arr.len()); + } + } + Ok(()) + } + + WorkflowCommands::Get { slug } => { + let result = + workflows_api::get_workflow_detail_api_v1_workflows_slug_get(config, &slug) + .await + .context("Failed to get workflow")?; + + if json_output { + println!("{}", serde_json::to_string_pretty(&result)?); + } else { + println!("Workflow: {}", result.name); + println!("Slug: {}", result.slug); + println!("ID: {}", result.id); + println!("Description: {}", result.description); + if let Some(Some(tags)) = &result.tags { + println!("Tags: {}", tags.join(", ")); + } + if let Some(nodes) = &result.nodes { + println!("\nNodes: {}", nodes.len()); + for node in nodes { + println!(" - {} ({:?})", node.node_id, node.component_type); + } + } + if let Some(edges) = &result.edges { + println!("\nEdges: {}", edges.len()); + } + } + Ok(()) + } + + WorkflowCommands::Validate { yaml } => { + let yaml_content = std::fs::read_to_string(&yaml) + .with_context(|| format!("Failed to read {}", yaml))?; + + let req = ValidateDslIn::new(yaml_content); + + let result = workflows_api::validate_dsl_endpoint_api_v1_workflows_validate_dsl_post( + config, req, + ) + .await + .context("Failed to validate DSL")?; + + if json_output { + println!("{}", serde_json::to_string_pretty(&result)?); + } else { + let valid = result["valid"].as_bool().unwrap_or(false); + if valid { + println!("Valid DSL"); + println!(" Nodes: {}", result["node_count"].as_i64().unwrap_or(0)); + println!(" Edges: {}", result["edge_count"].as_i64().unwrap_or(0)); + } else { + println!("Invalid DSL"); + if let Some(errors) = result["errors"].as_array() { + for err in errors { + println!(" - {}", err.as_str().unwrap_or("?")); + } + } + } + } + Ok(()) + } + + WorkflowCommands::Delete { slug, force } => { + if !force { + eprint!("Delete workflow '{}'? [y/N] ", slug); + let mut input = String::new(); + std::io::stdin().read_line(&mut input)?; + if !input.trim().eq_ignore_ascii_case("y") { + println!("Cancelled"); + return Ok(()); + } + } + + workflows_api::delete_workflow_api_v1_workflows_slug_delete(config, &slug) + .await + .context("Failed to delete workflow")?; + + println!("Deleted workflow: {}", slug); + Ok(()) + } + } +} + +async fn run_node_types(config: &Configuration, json_output: bool) -> Result<()> { + let result = workflows_api::list_node_types_api_v1_workflows_node_types_get(config) + .await + .context("Failed to list node types")?; + + if json_output { + println!("{}", serde_json::to_string_pretty(&result)?); + } else { + if let Some(obj) = result.as_object() { + println!("{:<25} {:<15} DESCRIPTION", "TYPE", "CATEGORY"); + println!("{}", "-".repeat(80)); + for (type_name, spec) in obj { + println!( + "{:<25} {:<15} {}", + type_name, + spec["category"].as_str().unwrap_or("-"), + truncate(spec["description"].as_str().unwrap_or("-"), 40), + ); + } + println!("\nTotal: {} node types", obj.len()); + } + } + Ok(()) +} + +fn truncate(s: &str, max: usize) -> String { + if s.len() <= max { + s.to_string() + } else { + format!("{}...", &s[..max - 3]) + } +} diff --git a/src/commands/init/agentgateway.rs b/src/commands/init/agentgateway.rs new file mode 100644 index 0000000..d8edfda --- /dev/null +++ b/src/commands/init/agentgateway.rs @@ -0,0 +1,573 @@ +//! agentgateway bootstrap for `plit init`. +//! +//! Creates the agentgateway config directory structure, generates an ES256 +//! keypair, writes config fragments, and shell scripts. The agentgateway +//! binary itself is NOT downloaded here — it is bundled in the Docker image +//! or downloaded separately for bare-metal installs. + +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, bail}; +use tokio::process::Command; + +use super::config; +use super::prompts::UserInputs; +use crate::output; + +/// Result of agentgateway bootstrap — returned to caller for .env writing. +pub struct AgentgatewaySetup { + /// Absolute path to the agentgateway directory + pub agw_dir: PathBuf, + /// PEM-encoded ES256 private key (for JWT signing) + pub jwt_private_key: String, +} + +/// Top-level directory for agentgateway config. +pub fn agentgateway_dir() -> Result { + let base = dirs::config_dir().context("Could not determine config directory")?; + Ok(base.join("agentgateway")) +} + +/// Bootstrap agentgateway config structure + ES256 keypair. +pub async fn bootstrap(inputs: &UserInputs) -> Result { + let agw_dir = agentgateway_dir()?; + + // Create directory structure + let dirs_to_create = [ + "bin", + "config.d/jwt", + "config.d/listeners", + "config.d/backends", + "config.d/rules", + "config.d/policies", + "config.d/mcp_servers", + "keys", + ]; + for d in &dirs_to_create { + let path = agw_dir.join(d); + std::fs::create_dir_all(&path) + .with_context(|| format!("Failed to create directory: {}", path.display()))?; + } + output::status(" * Created directory structure"); + + // Generate ES256 keypair via Python (cryptography is in the venv) + let (private_pem, jwks_json) = generate_es256_keypair().await?; + output::status(" * Generated ES256 keypair"); + + // Write JWKS public key + let jwks_path = agw_dir.join("config.d/jwt/jwks.json"); + std::fs::write(&jwks_path, &jwks_json) + .with_context(|| format!("Failed to write {}", jwks_path.display()))?; + + // Write config fragments + write_base_yaml(&agw_dir)?; + write_llm_listener_yaml(&agw_dir)?; + write_mcp_listener_yaml(&agw_dir)?; + write_rules(&agw_dir)?; + write_rate_limits(&agw_dir)?; + output::status(" * Wrote config fragments"); + + // Write shell scripts + write_assemble_config_sh(&agw_dir)?; + write_start_sh(&agw_dir)?; + write_decrypt_keys_py(&agw_dir)?; + output::status(" * Wrote shell scripts"); + + // Write initial provider + model from LLM choice (if applicable) + write_initial_provider(inputs, &agw_dir)?; + + Ok(AgentgatewaySetup { + agw_dir, + jwt_private_key: private_pem, + }) +} + +/// Generate an ES256 (P-256) keypair using Python's cryptography library. +/// Returns (private_key_pem, jwks_json). +async fn generate_es256_keypair() -> Result<(String, String)> { + let venv_dir = config::venv_dir()?; + let python = venv_dir.join("bin").join("python"); + + let script = r#" +import json +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.backends import default_backend +import base64 + +# Generate P-256 key +private_key = ec.generate_private_key(ec.SECP256R1(), default_backend()) +public_key = private_key.public_key() + +# PEM private key +private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption() +).decode() + +# JWKS public key +public_numbers = public_key.public_numbers() +x_bytes = public_numbers.x.to_bytes(32, 'big') +y_bytes = public_numbers.y.to_bytes(32, 'big') + +def b64url(b): + return base64.urlsafe_b64encode(b).rstrip(b'=').decode() + +jwks = { + "keys": [{ + "kty": "EC", + "crv": "P-256", + "x": b64url(x_bytes), + "y": b64url(y_bytes), + "use": "sig", + "alg": "ES256", + "kid": "pipelit-001" + }] +} + +print(json.dumps({"private_pem": private_pem, "jwks": jwks})) +"#; + + let output = Command::new(&python) + .args(["-c", script]) + .output() + .await + .context("Failed to run Python for ES256 keygen")?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + bail!("ES256 keygen failed: {}", stderr); + } + + let stdout = String::from_utf8(output.stdout).context("Invalid UTF-8 from keygen")?; + let parsed: serde_json::Value = + serde_json::from_str(&stdout).context("Failed to parse keygen output")?; + + let private_pem = parsed["private_pem"] + .as_str() + .context("Missing private_pem")? + .to_string(); + let jwks = serde_json::to_string_pretty(&parsed["jwks"]).context("Failed to format JWKS")?; + + Ok((private_pem, jwks)) +} + +// --- Config fragment writers --- + +fn write_base_yaml(agw_dir: &Path) -> Result<()> { + let content = "\ +config: + adminAddr: \"0.0.0.0:15000\" +"; + let path = agw_dir.join("config.d/base.yaml"); + std::fs::write(&path, content).with_context(|| format!("Failed to write {}", path.display())) +} + +fn write_llm_listener_yaml(agw_dir: &Path) -> Result<()> { + // The JWKS path must be absolute. In Docker it will be under /root/.config/agentgateway/ + // For bare-metal it will be under ~/.config/agentgateway/ + let jwks_path = agw_dir.join("config.d/jwt/jwks.json"); + let content = format!( + "\ +port: 4000 +listeners: +- name: llm + policies: + jwtAuth: + mode: strict + issuer: pipelit + audiences: [agentgateway] + jwks: + file: \"{jwks_path}\" + jwtValidationOptions: + requiredClaims: [] + routes: [] +", + jwks_path = jwks_path.display() + ); + let path = agw_dir.join("config.d/listeners/llm.yaml"); + std::fs::write(&path, &content).with_context(|| format!("Failed to write {}", path.display())) +} + +fn write_mcp_listener_yaml(agw_dir: &Path) -> Result<()> { + let content = "\ +port: 3000 +listeners: +- name: mcp + routes: + - policies: + cors: + allowOrigins: ['*'] + allowHeaders: [mcp-protocol-version, content-type, cache-control] + exposeHeaders: [Mcp-Session-Id] + backends: [] +"; + let path = agw_dir.join("config.d/listeners/mcp.yaml"); + std::fs::write(&path, content).with_context(|| format!("Failed to write {}", path.display())) +} + +fn write_rules(agw_dir: &Path) -> Result<()> { + let admin = "- 'jwt.role == \"admin\"'\n"; + let user = "- 'jwt.role == \"user\"'\n"; + + let admin_path = agw_dir.join("config.d/rules/admin.yaml"); + std::fs::write(&admin_path, admin) + .with_context(|| format!("Failed to write {}", admin_path.display()))?; + + let user_path = agw_dir.join("config.d/rules/user.yaml"); + std::fs::write(&user_path, user) + .with_context(|| format!("Failed to write {}", user_path.display())) +} + +/// Write the default local rate-limit policy fragment. +/// +/// assemble-config.sh injects these RateLimitSpec entries into EVERY LLM +/// route as `policies.localRateLimit` (schema verified against agentgateway +/// v1.0.1: `schema/config.json` + `examples/ratelimiting/local/config.yaml` +/// at that tag). The defaults are deliberately generous — a guard against +/// runaway autonomous agent loops, not a throttle on legitimate long runs. +/// +/// The file is only written if absent so operator tuning survives re-init. +/// Deleting the file (and reassembling) disables rate limiting entirely. +fn write_rate_limits(agw_dir: &Path) -> Result<()> { + let path = agw_dir.join("config.d/policies/rate-limit.yaml"); + if path.exists() { + return Ok(()); + } + let content = "\ +# Local rate limits applied to every LLM route (token bucket per route, +# state local to the gateway). Generous starter defaults: stop runaway +# agent loops without disturbing long legitimate agent runs. +# +# Schema (agentgateway v1.0.1 RateLimitSpec): +# maxTokens - bucket capacity (burst) +# tokensPerFill - added every fillInterval (sustained rate) +# fillInterval - duration string, e.g. 60s +# type - 'requests' (HTTP requests) or 'tokens' (LLM tokens) +# +# Tune per install; delete this file to disable rate limiting. +- maxTokens: 600 + tokensPerFill: 300 + fillInterval: 60s + type: requests +- maxTokens: 400000 + tokensPerFill: 200000 + fillInterval: 60s + type: tokens +"; + std::fs::write(&path, content).with_context(|| format!("Failed to write {}", path.display())) +} + +// --- Shell script writers --- + +fn write_assemble_config_sh(agw_dir: &Path) -> Result<()> { + let content = include_str!("agentgateway_scripts/assemble-config.sh"); + let path = agw_dir.join("assemble-config.sh"); + std::fs::write(&path, content) + .with_context(|| format!("Failed to write {}", path.display()))?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))?; + } + Ok(()) +} + +fn write_start_sh(agw_dir: &Path) -> Result<()> { + let content = include_str!("agentgateway_scripts/start.sh"); + let path = agw_dir.join("start.sh"); + std::fs::write(&path, content) + .with_context(|| format!("Failed to write {}", path.display()))?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))?; + } + Ok(()) +} + +fn write_decrypt_keys_py(agw_dir: &Path) -> Result<()> { + let content = include_str!("agentgateway_scripts/decrypt_keys.py"); + let path = agw_dir.join("decrypt_keys.py"); + std::fs::write(&path, content) + .with_context(|| format!("Failed to write {}", path.display()))?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))?; + } + Ok(()) +} + +// --- Initial provider/model from LLM choice --- + +fn write_initial_provider(inputs: &UserInputs, agw_dir: &Path) -> Result<()> { + // Map LLM provider choice to agentgateway provider config + let (provider_name, provider_type, host, path_override) = match inputs.llm_provider.as_str() { + "openai" => ("openai", "openAI", "https://api.openai.com", "/v1/"), + "anthropic" => ( + "anthropic", + "anthropic", + "https://api.anthropic.com", + "/v1/", + ), + "gemini" => ( + "gemini", + "gemini", + "https://generativelanguage.googleapis.com", + "/", + ), + "ollama" => { + let base = if inputs.llm_base_url.is_empty() { + "http://localhost:11434" + } else { + &inputs.llm_base_url + }; + // Ollama doesn't use API keys, skip encrypted key + write_provider_yaml(agw_dir, "ollama", "openAI", base, "/v1/")?; + write_model_yaml(agw_dir, "ollama", &inputs.llm_model)?; + output::status(&format!( + " * Created initial provider: ollama/{}", + model_slug(&inputs.llm_model) + )); + return Ok(()); + } + "openai-compatible" => { + if inputs.llm_base_url.is_empty() { + bail!("Base URL is required for openai-compatible providers"); + } + // Derive a provider name from the base URL hostname + let name = provider_name_from_url(&inputs.llm_base_url); + let path = extract_path(&inputs.llm_base_url, "openai-compatible"); + write_provider_yaml(agw_dir, &name, "openAI", &inputs.llm_base_url, &path)?; + write_model_yaml(agw_dir, &name, &inputs.llm_model)?; + write_encrypted_key(agw_dir, &name, &inputs.llm_api_key)?; + output::status(&format!( + " * Created initial provider: {}/{}", + name, + model_slug(&inputs.llm_model) + )); + return Ok(()); + } + _ => return Ok(()), // Unknown provider, skip + }; + + write_provider_yaml(agw_dir, provider_name, provider_type, host, path_override)?; + write_model_yaml(agw_dir, provider_name, &inputs.llm_model)?; + if !inputs.llm_api_key.is_empty() { + write_encrypted_key(agw_dir, provider_name, &inputs.llm_api_key)?; + } + output::status(&format!( + " * Created initial provider: {}/{}", + provider_name, + model_slug(&inputs.llm_model) + )); + + Ok(()) +} + +/// The agentgateway route name (`"-"`) that +/// `write_initial_provider` + `assemble-config.sh` produce for the initial +/// provider/model. plit init passes this to `apply-fixture --backend-route` +/// so pipelit records it on the default model node and its proxied LLM calls +/// hit the matching gateway route. Returns `None` for providers that create no +/// route. NOTE: the provider arm here must stay in sync with +/// `write_initial_provider` above. +pub fn initial_route_name(inputs: &UserInputs) -> Option { + let provider = match inputs.llm_provider.as_str() { + "openai" => "openai".to_string(), + "anthropic" => "anthropic".to_string(), + "gemini" => "gemini".to_string(), + "ollama" => "ollama".to_string(), + "openai-compatible" => { + if inputs.llm_base_url.is_empty() { + return None; + } + provider_name_from_url(&inputs.llm_base_url) + } + _ => return None, + }; + Some(format!("{}-{}", provider, model_slug(&inputs.llm_model))) +} + +fn write_provider_yaml( + agw_dir: &Path, + name: &str, + provider_type: &str, + host: &str, + path_override: &str, +) -> Result<()> { + let provider_dir = agw_dir.join("config.d/backends").join(name); + std::fs::create_dir_all(&provider_dir)?; + + // Build env var name from provider name + let env_var = format!("{}_API_KEY", name.to_uppercase().replace('-', "_")); + + // Extract just the hostname for hostOverride and SNI + let hostname = extract_host(host); + + // Emit backendTLS ONLY for TLS upstreams. agentgateway treats the mere + // presence of the policy (even {}) as "speak TLS to the upstream", which + // breaks plain-http backends (e.g. local Ollama or a LAN Qwen box) with + // a TLS InvalidContentType error. Only an explicit http:// scheme + // disables it; https:// (and scheme-less hosts) keep TLS on. + let backend_tls = if host.starts_with("http://") { + "" + } else { + "backendTLS: {}\n" + }; + + let content = format!( + "\ +provider: + {provider_type}: + model: \"\" +hostOverride: \"{hostname}\" +pathOverride: \"{path_override}\" +backendAuth: + key: + env: \"{env_var}\" +{backend_tls}", + provider_type = provider_type, + hostname = hostname, + path_override = path_override, + env_var = env_var, + backend_tls = backend_tls, + ); + + let path = provider_dir.join("_provider.yaml"); + std::fs::write(&path, content).with_context(|| format!("Failed to write {}", path.display())) +} + +fn write_model_yaml(agw_dir: &Path, provider_name: &str, model_name: &str) -> Result<()> { + let slug = model_slug(model_name); + let provider_dir = agw_dir.join("config.d/backends").join(provider_name); + std::fs::create_dir_all(&provider_dir)?; + + let content = format!("model: \"{}\"\n", model_name); + let path = provider_dir.join(format!("{}.yaml", slug)); + std::fs::write(&path, content).with_context(|| format!("Failed to write {}", path.display())) +} + +fn write_encrypted_key(agw_dir: &Path, provider_name: &str, api_key: &str) -> Result<()> { + if api_key.is_empty() { + return Ok(()); + } + // Write the key as plaintext for now — Pipelit's provider API will + // encrypt it with Fernet when providers are managed via the UI. + // For init bootstrap, plaintext is fine since start.sh handles both. + let path = agw_dir.join("keys").join(format!("{}.key", provider_name)); + std::fs::write(&path, api_key) + .with_context(|| format!("Failed to write {}", path.display()))?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))?; + } + Ok(()) +} + +/// Derive a model slug from a model name (e.g., "zai-org-glm-4.7" -> "glm-4.7"). +/// Keeps the full name but replaces characters unsafe for filenames. +fn model_slug(model_name: &str) -> String { + model_name.replace(['/', ':', ' '], "-") +} + +/// Derive a provider name from a base URL (e.g., "https://api.venice.ai/api/v1" -> "venice"). +/// +/// The result is always a valid environment-variable identifier component +/// (lowercase `[a-z0-9_]`, never leading with a digit) so that IP-based and +/// local backends work — e.g. "http://192.168.0.73:8080/v1" -> "p_192_168_0_73", +/// "http://localhost:11434" -> "localhost". A bare octet like "0" used to leak +/// through and produce an invalid `0_API_KEY` env var, breaking gateway startup. +fn provider_name_from_url(url: &str) -> String { + // Host without scheme, path, or port. + let host = url + .trim_end_matches('/') + .split("://") + .nth(1) + .unwrap_or(url) + .split('/') + .next() + .unwrap_or("") + .split(':') + .next() + .unwrap_or(""); + + // Real multi-label hostnames -> second-level domain (api.openai.com -> openai). + // IPs and single-label hosts -> use the whole host (localhost, 192.168.0.73). + let labels: Vec<&str> = host.split('.').filter(|s| !s.is_empty()).collect(); + let is_ip = !labels.is_empty() + && host + .split('.') + .all(|l| !l.is_empty() && l.bytes().all(|b| b.is_ascii_digit())); + let raw = if is_ip || labels.len() < 2 { + host + } else { + labels[labels.len() - 2] + }; + + // Sanitize into a valid identifier component. + let mut name: String = raw + .to_lowercase() + .chars() + .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' }) + .collect(); + if name.is_empty() { + name = "custom".to_string(); + } + if name.starts_with(|c: char| c.is_ascii_digit()) { + name = format!("p_{name}"); + } + name +} + +/// Extract host:port from a URL. Adds default port if missing (443 for https, 80 for http). +fn extract_host(url: &str) -> String { + let is_https = url.starts_with("https://"); + let host_part = url + .trim_end_matches('/') + .split("://") + .nth(1) + .unwrap_or(url) + .split('/') + .next() + .unwrap_or(url); + if host_part.contains(':') { + host_part.to_string() + } else if is_https { + format!("{host_part}:443") + } else { + format!("{host_part}:80") + } +} + +/// Extract path from a URL (e.g., "https://api.venice.ai/api/v1" -> "/api/v1/"). +fn extract_path(url: &str, provider_type: &str) -> String { + let after_scheme = url.split("://").nth(1).unwrap_or(url); + let path = after_scheme + .find('/') + .map(|i| &after_scheme[i..]) + .unwrap_or("/"); + let path = path.trim_end_matches('/'); + // Append endpoint suffix based on provider type + match provider_type { + "anthropic" => { + if path.ends_with("/messages") { + format!("{path}/") + } else { + format!("{path}/messages") + } + } + _ => { + // openai, openai-compatible, glm, gemini + if path.ends_with("/chat/completions") { + format!("{path}/") + } else { + format!("{path}/chat/completions") + } + } + } +} +// cache bust 1775094575 diff --git a/src/commands/init/agentgateway_scripts/assemble-config.sh b/src/commands/init/agentgateway_scripts/assemble-config.sh new file mode 100755 index 0000000..758d371 --- /dev/null +++ b/src/commands/init/agentgateway_scripts/assemble-config.sh @@ -0,0 +1,182 @@ +#!/usr/bin/env bash +# Assemble config.yaml from config.d/ domain folders +# +# Structure: +# config.d/base.yaml → top-level config (admin, global) +# config.d/listeners/*.yaml → one file per listener (each is a bind) +# config.d/backends// → one subfolder per LLM provider +# _provider.yaml → shared config (host, path, auth, TLS) +# .yaml → one file per model (just "model: ") +# config.d/rules/*.yaml → CEL authorization rules (merged into backends) +# config.d/policies/rate-limit.yaml → localRateLimit specs applied to every LLM route (optional) +# config.d/mcp_servers/*.yaml → MCP server targets (injected into MCP listener) +# config.d/jwt/jwks.json → JWT public key (referenced by llm listener) +# +# Only providers with matching keys/.key are included. +# Output: config.yaml (atomic write via tmp + rename) + +set -euo pipefail +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +CONFIG_D="$SCRIPT_DIR/config.d" +KEYS_DIR="$SCRIPT_DIR/keys" +OUTPUT="$SCRIPT_DIR/config.yaml" +YQ="${YQ:-yq}" + +# --- Step 1: Base config --- +merged=$($YQ eval '.' "$CONFIG_D/base.yaml") + +# --- Step 2: Load authorization rules --- +# Collect all rules/*.yaml into a single array +rules="[]" +for f in "$CONFIG_D"/rules/*.yaml; do + [ -f "$f" ] || continue + rules=$(echo "$rules" | $YQ eval-all ' + select(fileIndex == 0) + select(fileIndex == 1) + ' - "$f") +done + +# --- Step 2b: Load local rate limit policy (optional) --- +# config.d/policies/rate-limit.yaml holds an array of agentgateway +# RateLimitSpec entries (schema verified against agentgateway v1.0.1: +# maxTokens / tokensPerFill / fillInterval / type: requests|tokens). +# They are applied to EVERY LLM route as policies.localRateLimit — a +# generous runaway-spend guard. Delete the file OR comment out its +# entries to disable entirely. +# +# Emit JSON with a `// []` fallback so that an empty, null, or +# comment-only file collapses to "[]" (comments do not survive JSON +# encoding). Without this, a comment-only file would echo the comments +# and inject `localRateLimit: null`, which agentgateway rejects at +# startup — bricking the hard-boot dependency. +rate_limit="[]" +if [ -f "$CONFIG_D/policies/rate-limit.yaml" ]; then + rate_limit=$($YQ eval '. // []' -o=json "$CONFIG_D/policies/rate-limit.yaml") +fi + +# --- Step 3: Add listeners as binds --- +for f in "$CONFIG_D"/listeners/*.yaml; do + [ -f "$f" ] || continue + merged=$(echo "$merged" | $YQ eval-all ' + select(fileIndex == 0).binds += [select(fileIndex == 1)] + | select(fileIndex == 0) + ' - "$f") +done + +# --- Step 4: Inject backends into LLM listener routes --- +# Structure: config.d/backends//_provider.yaml + .yaml +# Each model file becomes a separate route, inheriting from _provider.yaml +# Only included if keys/.key exists +included=0 +skipped=0 +for provider_dir in "$CONFIG_D"/backends/*/; do + [ -d "$provider_dir" ] || continue + provider="$(basename "$provider_dir")" + provider_file="$provider_dir/_provider.yaml" + + # Skip if no _provider.yaml + if [ ! -f "$provider_file" ]; then + echo " ! $provider: missing _provider.yaml, skipped" + continue + fi + + # Skip if no matching key file + if [ ! -d "$KEYS_DIR" ] || [ ! -f "$KEYS_DIR/${provider}.key" ]; then + echo " - $provider (no key, skipped)" + skipped=$((skipped + 1)) + continue + fi + + # Read provider config + provider_config=$($YQ eval '.' "$provider_file") + + # Process each model file (skip _provider.yaml) + for model_file in "$provider_dir"/*.yaml; do + [ -f "$model_file" ] || continue + [ "$(basename "$model_file")" = "_provider.yaml" ] && continue + + model_slug="$(basename "$model_file" .yaml)" + model_name=$($YQ eval '.model' "$model_file") + route_name="${provider}-${model_slug}" + + # Build the full route by merging provider + model + route=$($YQ eval -n "{ + \"name\": \"${route_name}-route\", + \"matches\": [{\"path\": {\"pathPrefix\": \"/${route_name}/\"}}], + \"policies\": { + \"authorization\": {\"rules\": []}, + \"backendAuth\": $(echo "$provider_config" | $YQ eval '.backendAuth' -o=json -), + \"backendTLS\": $(echo "$provider_config" | $YQ eval '.backendTLS // {}' -o=json -) + }, + \"backends\": [{ + \"ai\": { + \"provider\": $(echo "$provider_config" | $YQ eval '.provider' -o=json -), + \"name\": \"${route_name}\" + $(echo "$provider_config" | $YQ eval 'select(.hostOverride) | ", \"hostOverride\": \"" + .hostOverride + "\""' -) + $(echo "$provider_config" | $YQ eval 'select(.pathOverride) | ", \"pathOverride\": \"" + .pathOverride + "\""' -) + } + }] + }") + + # Include backendTLS ONLY when the provider fragment declares it. + # agentgateway treats the mere presence of the policy (even {}) as + # "speak TLS to the upstream", which breaks plain-http backends + # (e.g. a LAN Qwen box or local Ollama) with InvalidContentType. + if [ "$(echo "$provider_config" | $YQ eval 'has("backendTLS")' -)" != "true" ]; then + route=$(echo "$route" | $YQ eval 'del(.policies.backendTLS)' -) + fi + + # Set the model override ONLY when the model file specifies a real + # upstream model id. A missing/null model means pass-through: + # agentgateway forwards the caller's requested model unchanged. Writing + # a "null" (or empty) override would ship literal "null" upstream and + # 404. (agentgateway provider..model is a hard outbound override.) + if [ -n "$model_name" ] && [ "$model_name" != "null" ]; then + route=$(echo "$route" | $YQ eval ".backends[0].ai.provider[][\"model\"] = \"${model_name}\"" -) + fi + + # Inject authorization rules + route=$(echo "$route" | $YQ eval-all ' + select(fileIndex == 0).policies.authorization.rules = select(fileIndex == 1) + | select(fileIndex == 0) + ' - <(echo "$rules")) + + # Inject local rate limits (per-route token bucket, local state). + # Skipped when config.d/policies/rate-limit.yaml is absent or empty, + # so installs without the fragment keep the previous behaviour. + if [ -n "$rate_limit" ] && [ "$rate_limit" != "[]" ] && [ "$rate_limit" != "null" ]; then + route=$(echo "$route" | $YQ eval-all ' + select(fileIndex == 0).policies.localRateLimit = select(fileIndex == 1) + | select(fileIndex == 0) + ' - <(echo "$rate_limit")) + fi + + # Append to LLM listener routes + merged=$(echo "$merged" | $YQ eval-all ' + (select(fileIndex == 0).binds[] | select(.listeners[].name == "llm")).listeners[0].routes += [select(fileIndex == 1)] + | select(fileIndex == 0) + ' - <(echo "$route")) + + echo " + ${provider}/${model_slug} (model: ${model_name})" + included=$((included + 1)) + done +done + +# --- Step 5: Inject MCP server targets into MCP listener --- +mcp_count=0 +for f in "$CONFIG_D"/mcp_servers/*.yaml; do + [ -f "$f" ] || continue + mcp_name="$(basename "$f" .yaml)" + + merged=$(echo "$merged" | $YQ eval-all ' + (select(fileIndex == 0).binds[] | select(.listeners[].name == "mcp")).listeners[0].routes[0].backends += [select(fileIndex == 1)] + | select(fileIndex == 0) + ' - "$f") + + echo " + mcp: $mcp_name" + mcp_count=$((mcp_count + 1)) +done + +# --- Step 6: Atomic write --- +echo "$merged" > "${OUTPUT}.tmp" +mv "${OUTPUT}.tmp" "$OUTPUT" +echo "Assembled config.yaml ($included backends, $skipped skipped, $mcp_count mcp servers)" diff --git a/src/commands/init/agentgateway_scripts/decrypt_keys.py b/src/commands/init/agentgateway_scripts/decrypt_keys.py new file mode 100755 index 0000000..7376cf3 --- /dev/null +++ b/src/commands/init/agentgateway_scripts/decrypt_keys.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Decrypt Fernet-encrypted .key files and print NAME=VALUE pairs. + +Usage: FIELD_ENCRYPTION_KEY=xxx python3 decrypt_keys.py /path/to/keys/ + +Called by start.sh before launching agentgateway. +Outputs lines like: VENICE_API_KEY=sk-actual-key +These are eval'd by start.sh to export as env vars. + +If FIELD_ENCRYPTION_KEY is not set, reads key files as plaintext +(backwards compatible with pre-encryption key files). +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + +from cryptography.fernet import Fernet + + +def _name_to_env_var(stem: str) -> str: + """Convert a key file stem to env var name. + + Examples: + venice -> VENICE_API_KEY + openai -> OPENAI_API_KEY + my-model -> MY_MODEL_API_KEY + """ + return stem.upper().replace("-", "_").replace(".", "_") + "_API_KEY" + + +def main() -> None: + keys_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("keys") + + if not keys_dir.is_dir(): + return + + enc_key = os.environ.get("FIELD_ENCRYPTION_KEY", "") + + if not enc_key: + print( + "Warning: FIELD_ENCRYPTION_KEY not set, reading keys as plaintext", + file=sys.stderr, + ) + for kf in sorted(keys_dir.glob("*.key")): + name = _name_to_env_var(kf.stem) + print(f"export {name}={kf.read_text().strip()}") + return + + try: + fernet = Fernet(enc_key.encode()) + except Exception: + # Invalid Fernet key — fall back to plaintext for all files + print("Warning: FIELD_ENCRYPTION_KEY is invalid, reading keys as plaintext", file=sys.stderr) + for kf in sorted(keys_dir.glob("*.key")): + name = _name_to_env_var(kf.stem) + print(f"export {name}={kf.read_text().strip()}") + return + + for kf in sorted(keys_dir.glob("*.key")): + name = _name_to_env_var(kf.stem) + try: + decrypted = fernet.decrypt(kf.read_bytes()).decode() + print(f"export {name}={decrypted}") + except Exception as e: + # Fall back to reading as plaintext + plaintext = kf.read_text().strip() + if plaintext: + print(f"export {name}={plaintext}") + print(f"Warning: Failed to decrypt {kf.name}, using as plaintext: {e}", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/src/commands/init/agentgateway_scripts/start.sh b/src/commands/init/agentgateway_scripts/start.sh new file mode 100755 index 0000000..78206cf --- /dev/null +++ b/src/commands/init/agentgateway_scripts/start.sh @@ -0,0 +1,69 @@ +#!/usr/bin/env bash +# Start agentgateway: +# 1. Decrypt + load API keys from keys/ folder +# 2. Assemble config.yaml from config.d/ fragments +# 3. Start agentgateway +# +# Key file convention: +# keys/venice.key → exports VENICE_API_KEY= +# keys/anthropic.key → exports ANTHROPIC_API_KEY= +# +# Config fragment convention: +# config.d/00-base.yaml → top-level config (admin, etc.) +# config.d/01-mcp.yaml → MCP listener (full bind) +# config.d/02-llm-base.yaml → LLM listener skeleton (JWT, empty routes) +# config.d/10-*.yaml → LLM route fragments (one per backend) + +set -euo pipefail +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +KEYS_DIR="$SCRIPT_DIR/keys" +YQ="${YQ:-yq}" + +# Find Python with cryptography (prefer Pipelit venv, fall back to system) +if [ -z "${PYTHON:-}" ]; then + for candidate in \ + "${PIPELIT_DIR:+$PIPELIT_DIR/.venv/bin/python3}" \ + "$HOME/.local/share/plit/venv/bin/python3" \ + "python3"; do + [ -n "$candidate" ] && [ -x "$candidate" ] && PYTHON="$candidate" && break + done +fi +PYTHON="${PYTHON:-python3}" + +# --- Step 1: Decrypt and load API keys --- +if [ -d "$KEYS_DIR" ] && ls "$KEYS_DIR"/*.key 1>/dev/null 2>&1; then + eval "$("${PYTHON:-python3}" "$SCRIPT_DIR/decrypt_keys.py" "$KEYS_DIR")" + echo "Loaded keys from $KEYS_DIR" +fi + +# --- Step 2: Assemble config from fragments --- +"$SCRIPT_DIR/assemble-config.sh" + +# --- Step 3: Assemble final config --- +# Convert YAML to JSON (workaround for serde_yaml untagged enum bug with -f). +# Then resolve env: references to inline values since -f doesn't support env. +$YQ eval -o=json '.' "$SCRIPT_DIR/config.yaml" > "$SCRIPT_DIR/config.tmp.json" + +# Replace {"env":"VAR_NAME"} with the actual env var value as a plain string +"${PYTHON}" -c " +import json, os, sys +with open('$SCRIPT_DIR/config.tmp.json') as f: + data = json.load(f) + +def resolve_env(obj): + if isinstance(obj, dict): + if set(obj.keys()) == {'env'} and isinstance(obj['env'], str): + return os.environ.get(obj['env'], '') + return {k: resolve_env(v) for k, v in obj.items()} + elif isinstance(obj, list): + return [resolve_env(v) for v in obj] + return obj + +data = resolve_env(data) +with open('$SCRIPT_DIR/config.json', 'w') as f: + json.dump(data, f) +" +rm -f "$SCRIPT_DIR/config.tmp.json" + +# --- Step 4: Start agentgateway --- +exec "$SCRIPT_DIR/bin/agentgateway" -f "$SCRIPT_DIR/config.json" "$@" diff --git a/src/commands/init/config.rs b/src/commands/init/config.rs index 566c35a..56bc407 100644 --- a/src/commands/init/config.rs +++ b/src/commands/init/config.rs @@ -223,6 +223,38 @@ pub fn write_gateway_config( Ok(()) } +/// Append agentgateway env vars to the existing .env file. +pub fn append_agentgateway_env(agw: &super::agentgateway::AgentgatewaySetup) -> Result<()> { + let path = dot_env_path()?; + + // Escape the PEM key for .env — replace newlines with literal \n + let escaped_key = agw.jwt_private_key.trim().replace('\n', "\\n"); + + let content = format!( + "\n\ +# agentgateway integration +AGENTGATEWAY_ENABLED=true +AGENTGATEWAY_URL=http://localhost:4000 +AGENTGATEWAY_DIR={agw_dir} +JWT_PRIVATE_KEY=\"{jwt_key}\" +", + agw_dir = agw.agw_dir.display(), + jwt_key = escaped_key, + ); + + let mut existing = std::fs::read_to_string(&path) + .with_context(|| format!("Failed to read {}", path.display()))?; + existing.push_str(&content); + std::fs::write(&path, &existing) + .with_context(|| format!("Failed to write {}", path.display()))?; + + output::status(&format!( + " * Appended agentgateway config to {}", + path.display() + )); + Ok(()) +} + pub fn write_managed_markers() -> Result<()> { let marker = ".plit-managed"; diff --git a/src/commands/init/install.rs b/src/commands/init/install.rs index 5c0e414..bd83673 100644 --- a/src/commands/init/install.rs +++ b/src/commands/init/install.rs @@ -201,6 +201,14 @@ pub async fn run_apply_fixture( .arg(&inputs.llm_base_url) .current_dir(pipelit_dir.join("platform")); + // Bridge the naming contract: tell pipelit the exact agentgateway route + // this provider/model will be reachable at, so the default node's + // backend_route matches what plit init creates (otherwise pipelit's + // proxied call 404s with "route not found"). + if let Some(route) = super::agentgateway::initial_route_name(inputs) { + cmd.arg("--backend-route").arg(route); + } + for (key, value) in &env_vars { cmd.env(key, value); } diff --git a/src/commands/init/mod.rs b/src/commands/init/mod.rs index 90e25cd..b6391a7 100644 --- a/src/commands/init/mod.rs +++ b/src/commands/init/mod.rs @@ -1,3 +1,4 @@ +mod agentgateway; pub(crate) mod config; mod docker; mod install; @@ -133,7 +134,13 @@ pub async fn run(args: InitArgs) -> Result<()> { // 19b. Drop management markers so uninstall can detect plit-managed dirs config::write_managed_markers()?; - // 20. Done + // 20. Bootstrap agentgateway config structure + ES256 keypair + output::status("Setting up agentgateway..."); + let agw_setup = agentgateway::bootstrap(&inputs).await?; + config::append_agentgateway_env(&agw_setup)?; + output::status(""); + + // 21. Done let config_path = config::config_json_path()?; let env_display = config::dot_env_path()?; let pipelit_display = config::pipelit_dir()?; diff --git a/src/commands/mod.rs b/src/commands/mod.rs index ddf7f09..b3e9643 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -1,3 +1,4 @@ +pub mod api; pub mod auth; pub mod chat; pub mod credentials; diff --git a/src/main.rs b/src/main.rs index f930be6..2ad7e06 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,5 @@ use clap::{Parser, Subcommand}; +use commands::api::ApiCommands; use commands::auth::AuthCommands; mod client; @@ -39,6 +40,10 @@ struct Cli { #[derive(Subcommand)] enum Commands { + /// Pipelit API operations (workflows, nodes, etc.) + #[command(subcommand)] + Api(ApiCommands), + /// Local gateway commands — chat, send, and listen via the generic adapter #[command(subcommand)] Local(LocalCommands), @@ -248,6 +253,8 @@ async fn main() -> anyhow::Result<()> { }; match cli.command { + Commands::Api(cmd) => commands::api::run(cmd, json_output).await, + Commands::Local(local_cmd) => match local_cmd { LocalCommands::Chat { credential_id,