# ai-skill-mcp/mcp_argo_pipeline architecture

> Architecture diagram of the GitHub repository ai-skill-mcp/mcp_argo_pipeline, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-08-03). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/ai-skill-mcp/mcp_argo_pipeline
- Repository: https://github.com/ai-skill-mcp/mcp_argo_pipeline

## Overview

#### Purpose and Entry Points
This repository exposes selected Argo `WorkflowTemplate`s as an MCP server, scoped to one named pipeline per process. The sole Python entry point is `server.py`; `start_serve.sh` is the server-oriented launcher that selects a pipeline and prepares dependencies. Clients use stdio locally or Streamable HTTP remotely.

Paths: `server.py`, `start_serve.sh`, `README.md`

#### Pipeline-Scoped Configuration
At startup, the process requires `MCP_PIPELINE` and resolves its pipeline definition: server name, kubeconfig, HTTP port, instructions, namespace, and permitted template names. Configuration file locations can be overridden with environment variables, separating deployment settings from source.

Paths: `common/pipeline_config.py`, `pipelines.yaml.example`, `common/runtime.py`

#### MCP Server and Transport Boundary
The server constructs the MCP-facing service for the chosen pipeline and selects `stdio` or `streamable-http` transport through `MCP_TRANSPORT`. Pipeline instructions become the MCP interaction context, while exposed operations are constrained to that pipeline’s configured templates.

Paths: `server.py`, `common/pipeline_server.py`, `common/runtime.py`

#### Authentication Boundary
HTTP transport requires Bearer authentication. Tokens are loaded per pipeline and map to identity metadata such as username and unit; this identity can be propagated into server-side request handling or workflow submission context. Stdio is intended for local MCP clients and does not use the remote HTTP authorization boundary.

Paths: `common/bearer_auth.py`, `auth.yaml.example`, `common/pipeline_server.py`

#### Argo Workflow Integration
The core execution boundary is the Argo client: MCP requests are translated into operations against Kubernetes-hosted Argo WorkflowTemplates. Only templates explicitly configured for the active pipeline should be reachable, preventing a pipeline server from becoming a general cluster workflow endpoint.

Paths: `common/argo_client.py`, `common/pipeline_server.py`, `common/pipeline_config.py`

#### Startup Validation and Runtime Dependencies
Before serving, startup checks validate kubeconfig access and configured template accessibility; `MCP_SKIP_STARTUP_CHECKS=1` bypasses this for troubleshooting. The service therefore depends materially on Kubernetes credentials, an accessible Kubernetes API/Argo installation, and the configured WorkflowTemplates.

Paths: `common/startup_checks.py`, `common/argo_client.py`, `common/runtime.py`

#### Deployment Model
Dependencies are managed with uv. `start_serve.sh` performs frozen synchronization unless disabled, then launches a single pipeline server, typically as an HTTP service on its configured port. Multiple pipelines are deployed as separate processes/endpoints rather than multiplexed through one server.

Paths: `start_serve.sh`, `pyproject.toml`, `uv.lock`

Main entry points: [`common/pipeline_server.py`](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/pipeline_server.py) (Pipeline MCP service), [`common/pipeline_config.py`](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/pipeline_config.py) (Pipeline configuration), [`common/argo_client.py`](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/argo_client.py) (Argo client), [`common/runtime.py`](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/runtime.py) (Runtime settings)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_startup["Startup & Configuration"]
  node_launcher["Server launcher<br/>shell launcher<br/>[start_serve.sh]"]
  node_runtime["Runtime settings<br/>environment resolver<br/>[runtime.py]"]
  node_pipeline_config["Pipeline configuration<br/>pipeline resolver<br/>[pipeline_config.py]"]
  node_pipeline_definitions["Pipeline definitions<br/>configuration example"]
  node_startup_checks["Startup checks<br/>validation<br/>[startup_checks.py]"]
  node_uv["uv dependencies<br/>dependency manager<br/>[pyproject.toml]"]
  node_uv_lock["Dependency lockfile<br/>[uv.lock]"]
end

subgraph group_mcp["Pipeline MCP Service"]
  node_server["Python entry point<br/>server entry<br/>[server.py]"]
  node_pipeline_service["Pipeline MCP service<br/>[pipeline_server.py]"]
  node_stdio(("Local stdio client<br/>MCP transport"))
  node_http["Streamable HTTP<br/>MCP transport"]
  node_bearer_auth["Bearer authentication<br/>HTTP auth<br/>[bearer_auth.py]"]
  node_auth_tokens["Token configuration<br/>configuration example<br/>[auth.yaml.example]"]
end

subgraph group_argo["Argo Execution Boundary"]
  node_argo_client["Argo client<br/>workflow adapter<br/>[argo_client.py]"]
  node_kubeconfig["Kubernetes credentials<br/>external credential"]
  node_kubernetes{{"Kubernetes API &amp; Argo<br/>external platform"}}
  node_workflow_templates["Configured WorkflowTemplates<br/>Argo resources"]
end

node_launcher -->|"syncs dependencies"| node_uv
node_uv_lock -->|"pins"| node_uv
node_launcher -->|"launches"| node_server
node_server -->|"reads"| node_runtime
node_runtime -->|"selects MCP_PIPELINE"| node_pipeline_config
node_pipeline_definitions -->|"defines pipelines"| node_pipeline_config
node_server -->|"runs before serving"| node_startup_checks
node_startup_checks -->|"validates access through"| node_argo_client
node_pipeline_config -->|"supplies kubeconfig and templates"| node_startup_checks
node_server -->|"constructs"| node_pipeline_service
node_pipeline_config -->|"scopes operations and instructions"| node_pipeline_service
node_runtime -->|"selects transport"| node_stdio
node_runtime -->|"selects transport"| node_http
node_stdio -->|"MCP requests"| node_pipeline_service
node_http -->|"requires"| node_bearer_auth
node_auth_tokens -->|"provides tokens"| node_bearer_auth
node_bearer_auth -->|"provides identity"| node_pipeline_service
node_http -->|"authorized MCP requests"| node_pipeline_service
node_pipeline_service -->|"submits permitted operations"| node_argo_client
node_pipeline_config -->|"supplies namespace and allowlist"| node_argo_client
node_kubeconfig -->|"authenticates"| node_argo_client
node_argo_client -->|"calls"| node_kubernetes
node_kubernetes -->|"hosts"| node_workflow_templates

click node_launcher "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/start_serve.sh"
click node_server "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/server.py"
click node_runtime "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/runtime.py"
click node_pipeline_config "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/pipeline_config.py"
click node_pipeline_definitions "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/pipelines.yaml.example"
click node_startup_checks "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/startup_checks.py"
click node_pipeline_service "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/pipeline_server.py"
click node_bearer_auth "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/bearer_auth.py"
click node_auth_tokens "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/auth.yaml.example"
click node_argo_client "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/argo_client.py"
click node_uv "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/pyproject.toml"
click node_uv_lock "https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/uv.lock"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_launcher,node_runtime,node_pipeline_config,node_pipeline_definitions,node_startup_checks,node_uv,node_uv_lock toneBlue
class node_server,node_pipeline_service,node_stdio,node_http,node_bearer_auth,node_auth_tokens toneAmber
class node_argo_client,node_kubeconfig,node_kubernetes,node_workflow_templates toneMint
```

## Components

### Startup & Configuration

- [Server launcher](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/start_serve.sh): shell launcher · Selects a pipeline, syncs dependencies, and starts one server process. · `start_serve.sh`
- [Runtime settings](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/runtime.py): environment resolver · Reads transport, pipeline, and startup-check settings. · `common/runtime.py`
- [Pipeline configuration](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/pipeline_config.py): pipeline resolver · Resolves the selected pipeline's namespace, templates, kubeconfig, port, and instructions. · `common/pipeline_config.py`
- [Pipeline definitions](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/pipelines.yaml.example): configuration example · `pipelines.yaml.example`
- [Startup checks](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/startup_checks.py): validation · Verifies kubeconfig and configured template accessibility unless skipped. · `common/startup_checks.py`
- [uv dependencies](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/pyproject.toml): dependency manager · Frozen synchronization uses the project manifest and lockfile. · `pyproject.toml`
- [Dependency lockfile](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/uv.lock): lockfile · `uv.lock`

### Pipeline MCP Service

- [Python entry point](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/server.py): server entry · `server.py`
- [Pipeline MCP service](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/pipeline_server.py): MCP service · Exposes only operations permitted for the active pipeline. · `common/pipeline_server.py`
- **Local stdio client**: MCP transport · Local clients bypass HTTP bearer authentication.
- **Streamable HTTP**: MCP transport · Remote MCP endpoint for the selected pipeline.
- [Bearer authentication](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/bearer_auth.py): HTTP auth · Maps per-pipeline tokens to identity metadata. · `common/bearer_auth.py`
- [Token configuration](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/auth.yaml.example): configuration example · `auth.yaml.example`

### Argo Execution Boundary

- [Argo client](https://github.com/ai-skill-mcp/mcp_argo_pipeline/blob/main/common/argo_client.py): workflow adapter · Translates MCP operations into Argo WorkflowTemplate operations. · `common/argo_client.py`
- **Kubernetes credentials**: external credential
- **Kubernetes API & Argo**: external platform
- **Configured WorkflowTemplates**: Argo resources · Only templates allowed by the active pipeline are reachable.

## Connections

- Server launcher → uv dependencies: syncs dependencies
- Dependency lockfile → uv dependencies: pins
- Server launcher → Python entry point: launches
- Python entry point → Runtime settings: reads
- Runtime settings → Pipeline configuration: selects MCP_PIPELINE
- Pipeline definitions → Pipeline configuration: defines pipelines
- Python entry point → Startup checks: runs before serving
- Startup checks → Argo client: validates access through
- Pipeline configuration → Startup checks: supplies kubeconfig and templates
- Python entry point → Pipeline MCP service: constructs
- Pipeline configuration → Pipeline MCP service: scopes operations and instructions
- Runtime settings → Local stdio client: selects transport
- Runtime settings → Streamable HTTP: selects transport
- Local stdio client → Pipeline MCP service: MCP requests
- Streamable HTTP → Bearer authentication: requires
- Token configuration → Bearer authentication: provides tokens
- Bearer authentication → Pipeline MCP service: provides identity
- Streamable HTTP → Pipeline MCP service: authorized MCP requests
- Pipeline MCP service → Argo client: submits permitted operations
- Pipeline configuration → Argo client: supplies namespace and allowlist
- Kubernetes credentials → Argo client: authenticates
- Argo client → Kubernetes API & Argo: calls
- Kubernetes API & Argo → Configured WorkflowTemplates: hosts