MCP 驱动 DevOps/SecOps：从工具链到智能自动化流水线

# src/mcp-cicd-plugin-framework.py
import json
import requests
from abc import ABC, abstractmethod
from typing import Dict, Any, List

class MCPCICDPlugin(ABC):
    """MCP CI/CD 插件抽象基类"""
    
    def __init__(self, mcp_server_url: str, api_key: str):
        self.mcp_server_url = mcp_server_url
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
    
    def call_mcp_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用 MCP 工具"""
        try:
            response = requests.post(
                f"{self.mcp_server_url}/api/v2/tools/call/{tool_name}",
                headers=self.headers,
                json=params,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    @abstractmethod
    def get_pipeline_context(self) -> Dict[str, Any]:
        """获取流水线上下文信息"""
        pass
    
    @abstractmethod
    def set_pipeline_variable(self, name: str, value: str) -> bool:
        """设置流水线变量"""
        pass
    
    @abstractmethod
    def upload_artifact(self, file_path: str, artifact_name: str) -> bool:
        """上传流水线产物"""
        pass
    
    def run_mcp_pipeline(self, pipeline_config: Dict[str, Any]) -> bool:
        """运行 MCP 驱动的流水线"""
        # 获取流水线上下文
        context = self.get_pipeline_context()
        
        # 合并上下文到流水线配置
        pipeline_config["context"] = context
        
        # 调用 MCP 流水线工具
        result = self.call_mcp_tool("pipeline_orchestrator", pipeline_config)
        
        if result.get("success", False):
            # 处理流水线结果
            self._handle_pipeline_result(result)
            return True
        else:
            # 处理流水线失败
            self._handle_pipeline_failure(result)
            return False
    
    def _handle_pipeline_result(self, result: Dict[str, Any]) -> None:
        """处理流水线成功结果"""
        # 设置流水线变量
        variables = result.get("variables", {})
        for name, value in variables.items():
            self.set_pipeline_variable(name, value)
        
        # 上传产物
        artifacts = result.get("artifacts", [])
        for artifact in artifacts:
            self.upload_artifact(artifact["file_path"], artifact["name"])
    
    def _handle_pipeline_failure(self, result: Dict[str, Any]) -> None:
        """处理流水线失败"""
        # 记录失败信息
        error = result.get("error", "Unknown error")
        print(f"MCP Pipeline failed: {error}")

class GitHubActionsMCPParser(MCPCICDPlugin):
    """GitHub Actions MCP 插件"""
    
    def get_pipeline_context(self) -> Dict[str, Any]:
        """获取 GitHub Actions 上下文"""
        import os
        return {
            "workflow": os.getenv("GITHUB_WORKFLOW"),
            "run_id": os.getenv("GITHUB_RUN_ID"),
            "repository": os.getenv("GITHUB_REPOSITORY"),
            "ref": os.getenv("GITHUB_REF"),
            "sha": os.getenv("GITHUB_SHA"),
            "event_name": os.getenv("GITHUB_EVENT_NAME"),
            "actor": os.getenv("GITHUB_ACTOR"),
            "workspace": os.getenv("GITHUB_WORKSPACE")
        }
    
    def set_pipeline_variable(self, name: str, value: str) -> bool:
        """设置 GitHub Actions 变量"""
        print(f"::set-output name={name}::{value}")
        return True
    
    def upload_artifact(self, file_path: str, artifact_name: str) -> bool:
        """上传 GitHub Actions 产物"""
        # 在 GitHub Actions 中，产物上传通常通过 actions/upload-artifact 动作完成
        # 这里仅作为示例
        print(f"Uploading artifact: {artifact_name} from {file_path}")
        return True

class JenkinsMCPParser(MCPCICDPlugin):
    """Jenkins MCP 插件"""
    
    def get_pipeline_context(self) -> Dict[str, Any]:
        """获取 Jenkins 上下文"""
        import os
        return {
            "job_name": os.getenv("JOB_NAME"),
            "build_number": os.getenv("BUILD_NUMBER"),
            "build_id": os.getenv("BUILD_ID"),
            "git_commit": os.getenv("GIT_COMMIT"),
            "git_branch": os.getenv("GIT_BRANCH"),
            "workspace": os.getenv("WORKSPACE")
        }
    
    def set_pipeline_variable(self, name: str, value: str) -> bool:
        """设置 Jenkins 变量"""
        print(f"##varset {name}={value}")
        return True
    
    def upload_artifact(self, file_path: str, artifact_name: str) -> bool:
        """上传 Jenkins 产物"""
        # 在 Jenkins 中，产物上传通常通过 archiveArtifacts 步骤完成
        # 这里仅作为示例
        print(f"Archiving artifact: {artifact_name} from {file_path}")
        return True

# 示例用法
if __name__ == "__main__":
    # 初始化 GitHub Actions MCP 插件
    mcp_plugin = GitHubActionsMCPParser(
        mcp_server_url="https://mcp-server.example.com",
        api_key="your-api-key"
    )
    
    # 定义流水线配置
    pipeline_config = {
        "name": "example-pipeline",
        "steps": [
            {
                "name": "code-quality-check",
                "tool": "code_quality_checker",
                "params": {
                    "repository": "${{ github.repository }}",
                    "ref": "${{ github.ref }}",
                    "sha": "${{ github.sha }}"
                }
            },
            {
                "name": "security-scan",
                "tool": "security_scanner",
                "params": {
                    "scan_type": "full",
                    "project_type": "python",
                    "source_path": "${{ github.workspace }}"
                }
            },
            {
                "name": "build-and-test",
                "tool": "build_and_test",
                "params": {
                    "source_path": "${{ github.workspace }}",
                    "language": "python",
                    "test_framework": "pytest"
                }
            }
        ]
    }
    
    # 运行 MCP 流水线
    success = mcp_plugin.run_mcp_pipeline(pipeline_config)
    
    if success:
        print("MCP Pipeline completed successfully")
    else:
        print("MCP Pipeline failed")
        exit(1)

# src/mcp-security-scan-orchestrator.py
import json
import yaml
import os
from typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum

class ScanType(Enum):
    """扫描类型枚举"""
    CODE = "code"
    CONTAINER = "container"
    DEPENDENCY = "dependency"
    INFRASTRUCTURE = "infrastructure"
    SECRET = "secret"

class RiskLevel(Enum):
    """风险等级枚举"""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ScanResult:
    """扫描结果数据类"""
    tool_name: str
    scan_type: ScanType
    risk_level: RiskLevel
    findings: List[Dict[str, Any]]
    scan_duration: float
    scan_date: str

class MCPSecurityScanOrchestrator:
    """MCP 安全扫描编排器"""
    
    def __init__(self, mcp_server_url: str, api_key: str):
        self.mcp_server_url = mcp_server_url
        self.api_key = api_key
        self.scan_configs = self._load_scan_configs()
    
    def _load_scan_configs(self) -> Dict[str, Any]:
        """加载扫描配置"""
        # 从文件加载配置
        config_file = "security_scan_configs.yaml"
        if os.path.exists(config_file):
            with open(config_file, "r") as f:
                return yaml.safe_load(f)
        
        # 默认配置
        return {
            "scan_tools": {
                "code": ["sonarqube", "eslint", "pylint"],
                "container": ["trivy", "clair", "anchore"],
                "dependency": ["snyk", "dependency-check", "ossindex"],
                "infrastructure": ["tfsec", "checkov"],
                "secret": ["gitleaks", "trufflehog"]
            },
            "risk_level_config": {
                "critical": {"threshold": 0, "action": "fail"},
                "high": {"threshold": 2, "action": "fail"},
                "medium": {"threshold": 5, "action": "warn"},
                "low": {"threshold": 10, "action": "info"}
            }
        }
    
    def _call_mcp_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用 MCP 工具"""
        import requests
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                f"{self.mcp_server_url}/api/v2/tools/call/{tool_name}",
                headers=headers,
                json=params,
                timeout=60
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"调用 MCP 工具失败: {e}")
            return {"success": False, "error": str(e)}
    
    def get_scan_tools(self, scan_type: ScanType) -> List[str]:
        """获取指定扫描类型的工具列表"""
        return self.scan_configs["scan_tools"].get(scan_type.value, [])
    
    def run_scan(self, scan_type: ScanType, params: Dict[str, Any]) -> List[ScanResult]:
        """运行安全扫描"""
        scan_results = []
        
        # 获取扫描工具列表
        scan_tools = self.get_scan_tools(scan_type)
        
        # 依次调用扫描工具
        for tool in scan_tools:
            # 调用 MCP 扫描工具
            tool_params = params.copy()
            tool_params["scan_type"] = scan_type.value
            
            result = self._call_mcp_tool(tool, tool_params)
            
            if result.get("success", False):
                # 处理扫描结果
                scan_result = self._parse_scan_result(tool, scan_type, result)
                scan_results.append(scan_result)
            else:
                # 处理扫描失败
                print(f"扫描工具 {tool} 失败: {result.get('error', 'Unknown error')}")
        
        return scan_results
    
    def _parse_scan_result(self, tool_name: str, scan_type: ScanType, result: Dict[str, Any]) -> ScanResult:
        """解析扫描结果"""
        findings = result.get("findings", [])
        scan_duration = result.get("scan_duration", 0.0)
        scan_date = result.get("scan_date", "")
        
        # 确定风险等级
        risk_level = self._determine_risk_level(findings)
        
        return ScanResult(
            tool_name=tool_name,
            scan_type=scan_type,
            risk_level=risk_level,
            findings=findings,
            scan_duration=scan_duration,
            scan_date=scan_date
        )
    
    def _determine_risk_level(self, findings: List[Dict[str, Any]]) -> RiskLevel:
        """根据发现的问题确定风险等级"""
        # 统计各风险等级的数量
        risk_counts = {
            "critical": 0,
            "high": 0,
            "medium": 0,
            "low": 0
        }
        
        for finding in findings:
            severity = finding.get("severity", "low").lower()
            if severity in risk_counts:
                risk_counts[severity] += 1
        
        # 确定最高风险等级
        if risk_counts["critical"] > 0:
            return RiskLevel.CRITICAL
        elif risk_counts["high"] > 0:
            return RiskLevel.HIGH
        elif risk_counts["medium"] > 0:
            return RiskLevel.MEDIUM
        else:
            return RiskLevel.LOW
    
    def aggregate_scan_results(self, scan_results: List[ScanResult]) -> Dict[str, Any]:
        """聚合扫描结果"""
        aggregated_findings = []
        total_duration = 0.0
        risk_counts = {
            "critical": 0,
            "high": 0,
            "medium": 0,
            "low": 0
        }
        
        # 聚合结果
        for result in scan_results:
            # 累加扫描时长
            total_duration += result.scan_duration
            
            # 聚合发现的问题
            aggregated_findings.extend(result.findings)
            
            # 统计风险等级
            risk_counts[result.risk_level.value] += 1
        
        # 确定整体风险等级
        overall_risk = RiskLevel.LOW
        if risk_counts["critical"] > 0:
            overall_risk = RiskLevel.CRITICAL
        elif risk_counts["high"] > 0:
            overall_risk = RiskLevel.HIGH
        elif risk_counts["medium"] > 0:
            overall_risk = RiskLevel.MEDIUM
        
        # 生成报告
        report = {
            "scan_summary": {
                "total_scans": len(scan_results),
                "total_duration": total_duration,
                "overall_risk": overall_risk.value,
                "risk_counts": risk_counts,
                "total_findings": len(aggregated_findings)
            },
            "scan_results": [{
                "tool_name": result.tool_name,
                "scan_type": result.scan_type.value,
                "risk_level": result.risk_level.value,
                "findings_count": len(result.findings),
                "scan_duration": result.scan_duration,
                "scan_date": result.scan_date
            } for result in scan_results],
            "aggregated_findings": aggregated_findings
        }
        
        return report
    
    def evaluate_risk(self, aggregated_results: Dict[str, Any]) -> Dict[str, Any]:
        """评估风险并生成建议"""
        scan_summary = aggregated_results["scan_summary"]
        overall_risk = scan_summary["overall_risk"]
        risk_counts = scan_summary["risk_counts"]
        
        # 根据风险等级和数量确定动作
        action = "pass"
        message = "安全扫描通过"
        
        for risk_level in ["critical", "high", "medium", "low"]:
            threshold = self.scan_configs["risk_level_config"][risk_level]["threshold"]
            risk_action = self.scan_configs["risk_level_config"][risk_level]["action"]
            
            if risk_counts[risk_level] > threshold:
                action = risk_action
                message = f"安全扫描 {risk_action}：发现 {risk_counts[risk_level]} 个 {risk_level} 风险"
                break
        
        return {
            "action": action,
            "message": message,
            "scan_summary": scan_summary,
            "risk_breakdown": risk_counts
        }
    
    def generate_scan_report(self, aggregated_results: Dict[str, Any], output_file: str = "security_scan_report.json") -> bool:
        """生成扫描报告"""
        try:
            with open(output_file, "w") as f:
                json.dump(aggregated_results, f, indent=2)
            print(f"扫描报告已保存到 {output_file}")
            return True
        except Exception as e:
            print(f"生成扫描报告失败: {e}")
            return False

# 示例用法
if __name__ == "__main__":
    # 初始化安全扫描编排器
    orchestrator = MCPSecurityScanOrchestrator(
        mcp_server_url="https://mcp-server.example.com",
        api_key="your-api-key"
    )
    
    # 运行代码扫描
    print("运行代码扫描...")
    code_scan_results = orchestrator.run_scan(
        ScanType.CODE,
        {
            "source_path": ".",
            "language": "python",
            "scan_depth": "full"
        }
    )
    
    # 运行依赖扫描
    print("运行依赖扫描...")
    dependency_scan_results = orchestrator.run_scan(
        ScanType.DEPENDENCY,
        {
            "source_path": ".",
            "language": "python",
            "package_manager": "pip"
        }
    )
    
    # 运行容器扫描
    print("运行容器扫描...")
    container_scan_results = orchestrator.run_scan(
        ScanType.CONTAINER,
        {
            "image": "your-docker-image:latest",
            "scan_type": "vulnerability"
        }
    )
    
    # 聚合所有扫描结果
    all_results = code_scan_results + dependency_scan_results + container_scan_results
    aggregated_results = orchestrator.aggregate_scan_results(all_results)
    
    # 评估风险
    risk_evaluation = orchestrator.evaluate_risk(aggregated_results)
    
    # 生成报告
    orchestrator.generate_scan_report(aggregated_results)
    
    # 输出结果
    print("\n安全扫描结果摘要:")
    print(f"整体风险等级: {aggregated_results['scan_summary']['overall_risk']}")
    print(f"风险计数: {aggregated_results['scan_summary']['risk_counts']}")
    print(f"评估结果: {risk_evaluation['action']} - {risk_evaluation['message']}")
    
    # 根据评估结果决定是否失败
    if risk_evaluation['action'] == 'fail':
        print("\n安全扫描失败，流水线终止")
        exit(1)
    else:
        print("\n安全扫描通过，流水线继续")
        exit(0)

// main.go
package main

import (
	"flag"
	"fmt"
	"os"
	"runtime"

	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
	// to ensure that exec-entrypoint and run can make use of them.
	_ "k8s.io/client-go/plugin/pkg/client/auth"

	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"

	mcpv1alpha1 "github.com/mcp-operator/mcp-operator/api/v1alpha1"
	"github.com/mcp-operator/mcp-operator/controllers"
	//+kubebuilder:scaffold:imports
)

var (
	scheme	= runtime.NewScheme()
	setupLog = ctrl.Log.WithName("setup")
)

func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))

	utilruntime.Must(mcpv1alpha1.AddToScheme(scheme))
	//+kubebuilder:scaffold:scheme
}

func main() {
	var metricsAddr string
	var enableLeaderElection bool
	var probeAddr string
	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
	flag.BoolVar(&enableLeaderElection, "leader-elect", false, "Enable leader election for controller manager.")
	opts := zap.Options{
		Development: true,
	}
	opts.BindFlags(flag.CommandLine)
	flag.Parse()

	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:                 scheme,
		MetricsBindAddress:     metricsAddr,
		Port:                   9443,
		HealthProbeBindAddress: probeAddr,
		LeaderElection:         enableLeaderElection,
		LeaderElectionID:       "mcp-operator.example.com",
		// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily
		// when the Manager ends. This requires the binary to immediately end when the
		// Manager is stopped, otherwise, this setting is unsafe. Setting this significantly
		// speeds up voluntary leader transitions as the new leader don't have to wait
		// LeaseDuration time first.
		//
		// In the default scaffold provided, the program ends immediately after
		// the manager stops, so would be fine to enable this option.
		// However, if you are doing or is intended to do any operation such as perform cleanups
		// after the manager stops then its usage might be unsafe.
		LeaderElectionReleaseOnCancel: true,
	})
	if err != nil {
		setupLog.Error(err, "unable to start manager")
		os.Exit(1)
	}

	// 初始化 MCP 客户端
	mcpClient, err := controllers.NewMCPClient()
	if err != nil {
		setupLog.Error(err, "unable to create MCP client")
		os.Exit(1)
	}

	// 初始化控制器
	if err = (&controllers.MCPToolReconciler{
		Client:    mgr.GetClient(),
		Scheme:    mgr.GetScheme(),
		MCPClient: mcpClient,
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "MCPTool")
		os.Exit(1)
	}

	if err = (&controllers.MCPToolInstanceReconciler{
		Client:    mgr.GetClient(),
		Scheme:    mgr.GetScheme(),
		MCPClient: mcpClient,
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "MCPToolInstance")
		os.Exit(1)
	}

	if err = (&controllers.MCPPipelineReconciler{
		Client:    mgr.GetClient(),
		Scheme:    mgr.GetScheme(),
		MCPClient: mcpClient,
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "MCPPipeline")
		os.Exit(1)
	}
	//+kubebuilder:scaffold:builder

	// 设置健康检查
	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		setupLog.Error(err, "unable to set up health check")
		os.Exit(1)
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		setupLog.Error(err, "unable to set up ready check")
		os.Exit(1)
	}

	setupLog.Info("starting manager")
	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		setupLog.Error(err, "problem running manager")
		os.Exit(1)
	}
}

// controllers/mcp_tool_reconciler.go
package controllers

import (
	"context"
	"fmt"

	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	mcpv1alpha1 "github.com/mcp-operator/mcp-operator/api/v1alpha1"
)

// MCPToolReconciler 用于协调 MCPTool 资源
type MCPToolReconciler struct {
	client.Client
	Log       logr.Logger
	Scheme    *runtime.Scheme
	MCPClient *MCPClient
}

//+kubebuilder:rbac:groups=mcp.example.com,resources=mcptools,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=mcp.example.com,resources=mcptools/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=mcp.example.com,resources=mcptools/finalizers,verbs=update

// Reconcile 协调 MCPTool 资源
func (r *MCPToolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := r.Log.WithValues("mcptool", req.NamespacedName)

	// 获取 MCPTool 资源
	var mcptool mcpv1alpha1.MCPTool
	if err := r.Get(ctx, req.NamespacedName, &mcptool); err != nil {
		if apierrors.IsNotFound(err) {
			// 资源已被删除，不需要处理
			return ctrl.Result{}, nil
		}
		log.Error(err, "无法获取 MCPTool 资源")
		return ctrl.Result{}, err
	}

	// 检查资源是否被标记为删除
	if !mcptool.DeletionTimestamp.IsZero() {
		// 处理资源删除逻辑
		return r.handleDeletion(ctx, &mcptool)
	}

	// 处理资源创建/更新逻辑
	return r.handleCreationOrUpdate(ctx, &mcptool)
}

// handleCreationOrUpdate 处理资源创建或更新
func (r *MCPToolReconciler) handleCreationOrUpdate(ctx context.Context, mcptool *mcpv1alpha1.MCPTool) (ctrl.Result, error) {
	log := r.Log.WithValues("mcptool", mcptool.Name)
	
	// 注册 MCP 工具到 MCP Server
	log.Info("注册 MCP 工具", "toolName", mcptool.Spec.Name)
	
	// 调用 MCP Server API 注册工具
	result, err := r.MCPClient.RegisterTool(mcptool.Spec)
	if err != nil {
		log.Error(err, "注册 MCP 工具失败")
		// 更新状态为失败
		mcptool.Status.Ready = corev1.ConditionFalse
		mcptool.Status.Message = fmt.Sprintf("注册工具失败: %v", err)
		if err := r.Status().Update(ctx, mcptool); err != nil {
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, err
	}
	
	// 更新状态为成功
	mcptool.Status.Ready = corev1.ConditionTrue
	mcptool.Status.Message = "MCP 工具注册成功"
	mcptool.Status.ToolID = result.ToolID
	if err := r.Status().Update(ctx, mcptool); err != nil {
		return ctrl.Result{}, err
	}
	
	log.Info("MCP 工具注册成功", "toolID", result.ToolID)
	return ctrl.Result{}, nil
}

// handleDeletion 处理资源删除
func (r *MCPToolReconciler) handleDeletion(ctx context.Context, mcptool *mcpv1alpha1.MCPTool) (ctrl.Result, error) {
	log := r.Log.WithValues("mcptool", mcptool.Name)
	
	// 从 MCP Server 注销工具
	log.Info("注销 MCP 工具", "toolID", mcptool.Status.ToolID)
	
	if mcptool.Status.ToolID != "" {
		if err := r.MCPClient.UnregisterTool(mcptool.Status.ToolID); err != nil {
			log.Error(err, "注销 MCP 工具失败")
			return ctrl.Result{}, err
		}
	}
	
	log.Info("MCP 工具注销成功")
	return ctrl.Result{}, nil
}

// SetupWithManager 设置控制器
func (r *MCPToolReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&mcpv1alpha1.MCPTool{}).
		Complete(r)
}

scan_tools:
  code: ["sonarqube", "eslint", "pylint"]
  container: ["trivy", "clair", "anchore"]
  dependency: ["snyk", "dependency-check", "ossindex"]
  infrastructure: ["tfsec", "checkov"]
  secret: ["gitleaks", "trufflehog"]

risk_level_config:
  critical:
    threshold: 0
    action: "fail"
  high:
    threshold: 2
    action: "fail"
  medium:
    threshold: 5
    action: "warn"
  low:
    threshold: 10
    action: "info"

scan_profiles:
  default:
    code: true
    container: true
    dependency: true
    infrastructure: false
    secret: true
  strict:
    code: true
    container: true
    dependency: true
    infrastructure: true
    secret: true
  minimal:
    code: true
    container: false
    dependency: true
    infrastructure: false
    secret: false

apiVersion: mcp.example.com/v1alpha1
kind: MCPTool
metadata:
  name: example-mcptool
  namespace: default
spec:
  name: "k8s_resource_manager"
  description: "Kubernetes 资源管理器，用于管理 Kubernetes 资源"
  parameters:
    type: "object"
    properties:
      resource_type:
        type: "string"
        description: "Kubernetes 资源类型，如 deployment、service 等"
        required: true
      namespace:
        type: "string"
        description: "Kubernetes 命名空间"
        required: false
      name:
        type: "string"
        description: "Kubernetes 资源名称"
        required: false
      action:
        type: "string"
        description: "要执行的操作，如 get、create、update、delete"
        required: true
      spec:
        type: "object"
        description: "Kubernetes 资源规格"
        required: false
  result:
    type: "object"
    properties:
      success:
        type: "boolean"
        description: "操作是否成功"
      message:
        type: "string"
        description: "操作结果消息"
      resource:
        type: "object"
        description: "Kubernetes 资源对象"
      error:
        type: "string"
        description: "错误信息"

name: "example-pipeline"
description: "示例 MCP 流水线"
steps:
  - name: "code-quality-check"
    tool: "code_quality_checker"
    params:
      repository: "${{ github.repository }}"
      ref: "${{ github.ref }}"
      sha: "${{ github.sha }}"
    on_failure: "continue"
    retry_count: 1
    retry_delay: 30
  
  - name: "security-scan"
    tool: "security_scanner"
    params:
      scan_type: "full"
      project_type: "python"
      source_path: "${{ github.workspace }}"
    on_failure: "fail"
  
  - name: "build-and-test"
    tool: "build_and_test"
    params:
      source_path: "${{ github.workspace }}"
      language: "python"
      test_framework: "pytest"
    on_failure: "fail"
  
  - name: "deploy-to-test"
    tool: "k8s_deployer"
    params:
      cluster: "test-cluster"
      namespace: "test"
      manifest_path: "${{ github.workspace }}/k8s/deployment.yaml"
      image_tag: "${{ github.sha }}"
    on_failure: "fail"
    depends_on: ["build-and-test"]
  
  - name: "run-integration-tests"
    tool: "integration_tester"
    params:
      test_url: "https://test.example.com"
      test_suite: "api"
    on_failure: "fail"
    depends_on: ["deploy-to-test"]
  
  - name: "deploy-to-production"
    tool: "k8s_deployer"
    params:
      cluster: "prod-cluster"
      namespace: "prod"
      manifest_path: "${{ github.workspace }}/k8s/deployment.yaml"
      image_tag: "${{ github.sha }}"
      rollout_strategy: "canary"
    on_failure: "manual"
    depends_on: ["run-integration-tests"]