Phần 6 — Chuyển dịch vai trò: Từ Coder đến AI Orchestrator trong Môi trường MCP 2026

package main

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/errgroup"
)

// AgentRole đại diện cho vai trò chuyên biệt của từng agent trong đàn.
type AgentRole string

const (
	RoleDatabase AgentRole = "DATABASE_AGENT"
	RoleBackend  AgentRole = "BACKEND_AGENT"
	RoleFrontend AgentRole = "FRONTEND_AGENT"
	RoleSecurity AgentRole = "SECURITY_AGENT"
)

// SwarmTask chứa gói dữ liệu công việc độc lập phân bổ cho một sub-agent.
type SwarmTask struct {
	ID      string        `json:"id"`
	Target  AgentRole     `json:"target"`
	Payload string        `json:"payload"`
	Timeout time.Duration `json:"timeout"`
}

// AgentHandoffArtifact là kết quả bàn giao bất biến do sub-agent tạo ra.
type AgentHandoffArtifact struct {
	TaskID    string    `json:"task_id"`
	Agent     AgentRole `json:"agent"`
	Result    string    `json:"result"`
	Status    string    `json:"status"`
	Timestamp time.Time `json:"timestamp"`
}

// SwarmOrchestrator quản lý việc điều phối và đồng bộ hóa kết quả từ các agent.
type SwarmOrchestrator struct {
	mu        sync.Mutex
	artifacts []AgentHandoffArtifact
}

func NewSwarmOrchestrator() *SwarmOrchestrator {
	return &SwarmOrchestrator{
		artifacts: make([]AgentHandoffArtifact, 0),
	}
}

// DispatchSwarm thực thi đồng thời các tác vụ với kiểm soát context độc lập.
func (o *SwarmOrchestrator) DispatchSwarm(parentCtx context.Context, tasks []SwarmTask) ([]AgentHandoffArtifact, error) {
	g, ctx := errgroup.WithContext(parentCtx)

	for _, task := range tasks {
		t := task
		g.Go(func() error {
			taskCtx, cancel := context.WithTimeout(ctx, t.Timeout)
			defer cancel()

			artifact, err := o.executeSubAgentTask(taskCtx, t)
			if err != nil {
				return fmt.Errorf("agent %s thất bại khi thực hiện tác vụ %s: %w", t.Target, t.ID, err)
			}

			o.mu.Lock()
			o.artifacts = append(o.artifacts, artifact)
			o.mu.Unlock()
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}

	return o.artifacts, nil
}

func (o *SwarmOrchestrator) executeSubAgentTask(ctx context.Context, task SwarmTask) (AgentHandoffArtifact, error) {
	select {
	case <-ctx.Done():
		return AgentHandoffArtifact{}, ctx.Err()
	default:
		var resultPayload string
		switch task.Target {
		case RoleDatabase:
			checksum := sha256.Sum256([]byte(task.Payload))
			resultPayload = fmt.Sprintf(`{"role":"%s","status":"DDL_MIGRATION_READY","schema_hash":"%x","bytes":%d}`,
				task.Target, checksum[:8], len(task.Payload))
		case RoleBackend:
			resultPayload = fmt.Sprintf(`{"role":"%s","status":"GRPC_HANDLERS_COMPILED","methods":["CreateAccount","GetAccountBalance"],"bytes":%d}`,
				task.Target, len(task.Payload))
		case RoleFrontend:
			resultPayload = fmt.Sprintf(`{"role":"%s","status":"REACT_COMPONENTS_BUNDLED","bundle_size":4200,"type_safe":true}`,
				task.Target)
		case RoleSecurity:
			hasAuth := strings.Contains(task.Payload, "JWT") || strings.Contains(task.Payload, "Bearer")
			resultPayload = fmt.Sprintf(`{"role":"%s","status":"SECURITY_AUDIT_PASSED","auth_verified":%t}`,
				task.Target, hasAuth)
		default:
			resultPayload = fmt.Sprintf(`{"role":"%s","status":"EXECUTED","bytes":%d}`, task.Target, len(task.Payload))
		}

		return AgentHandoffArtifact{
			TaskID:    task.ID,
			Agent:     task.Target,
			Result:    resultPayload,
			Status:    "SUCCESS",
			Timestamp: time.Now(),
		}, nil
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	orchestrator := NewSwarmOrchestrator()

	tasks := []SwarmTask{
		{ID: "task-01", Target: RoleDatabase, Payload: "CREATE TABLE ledgers (id UUID PRIMARY KEY, balance NUMERIC);", Timeout: 5 * time.Second},
		{ID: "task-02", Target: RoleBackend, Payload: "Implement LedgerService gRPC handlers with ACID transactions", Timeout: 5 * time.Second},
		{ID: "task-03", Target: RoleFrontend, Payload: "Synthesize LedgerDashboard React views with Tailwind CSS", Timeout: 5 * time.Second},
		{ID: "task-04", Target: RoleSecurity, Payload: "Audit gRPC LedgerService for JWT bearer token validation", Timeout: 5 * time.Second},
	}

	artifacts, err := orchestrator.DispatchSwarm(ctx, tasks)
	if err != nil {
		log.Fatalf("Điều phối swarm thất bại: %v", err)
	}

	fmt.Printf("=== Điều phối Swarm hoàn tất thành công (%d Artifacts) ===\n", len(artifacts))
	for _, art := range artifacts {
		fmt.Printf("[%s] Vai trò: %-15s -> Kết quả: %s\n", art.Status, art.Agent, art.Result)
	}
}

#!/usr/bin/env python3
"""
Máy chủ Model Context Protocol (MCP 2.0) chuẩn Production.
Hiện thực hóa chuẩn giao thức JSON-RPC 2.0 qua luồng Standard I/O.
Cung cấp công cụ kiểm tra schema, thực thi truy vấn sandbox và phân tích git blame.
"""

import sys
import json
import sqlite3
import subprocess
from dataclasses import dataclass, asdict
from typing import Any, Dict, List, Optional
from pathlib import Path

JSONRPC_VERSION = "2.0"

@dataclass
class ToolDefinition:
    name: str
    description: str
    inputSchema: Dict[str, Any]

class MCPServer:
    def __init__(self, db_path: str = ":memory:"):
        self.db_path = db_path
        self._init_sample_db()
        self.tools: Dict[str, ToolDefinition] = {
            "inspect_db_schema": ToolDefinition(
                name="inspect_db_schema",
                description="Trích xuất cấu trúc bảng, danh sách cột và kiểu dữ liệu trong cơ sở dữ liệu.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "table_name": {"type": "string", "description": "Tên bảng cần kiểm tra"}
                    },
                    "required": ["table_name"]
                }
            ),
            "execute_sandboxed_query": ToolDefinition(
                name="execute_sandboxed_query",
                description="Thực thi câu truy vấn SQL SELECT an toàn trong môi trường sandbox có giới hạn.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Câu lệnh SQL SELECT cần chạy"},
                        "limit": {"type": "integer", "description": "Số lượng bản ghi tối đa", "default": 20}
                    },
                    "required": ["query"]
                }
            ),
            "git_blame_analyzer": ToolDefinition(
                name="git_blame_analyzer",
                description="Phân tích tác giả commit và lịch sử chỉnh sửa trên phạm vi dòng code cụ thể.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "file_path": {"type": "string", "description": "Đường dẫn file trong kho Git"},
                        "start_line": {"type": "integer", "description": "Dòng bắt đầu"},
                        "end_line": {"type": "integer", "description": "Dòng kết thúc"}
                    },
                    "required": ["file_path"]
                }
            )
        }

    def _init_sample_db(self) -> None:
        """Khởi tạo cơ sở dữ liệu mẫu trong RAM để phục vụ truy vấn sandbox."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS accounts (
                    id TEXT PRIMARY KEY,
                    owner_email TEXT NOT NULL,
                    balance REAL NOT NULL DEFAULT 0.0,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );
            """)
            cursor.execute("""
                INSERT OR IGNORE INTO accounts (id, owner_email, balance)
                VALUES ('acc_01', 'architect@enterprise.corp', 125000.50);
            """)
            conn.commit()

    def handle_request(self, request_str: str) -> Optional[str]:
        """Xử lý yêu cầu JSON-RPC 2.0 từ luồng client."""
        try:
            req = json.loads(request_str)
        except json.JSONDecodeError as e:
            return self._build_error_response(None, -32700, f"Lỗi phân tích cú pháp JSON: {e}")

        req_id = req.get("id")
        method = req.get("method")
        params = req.get("params", {})

        if not method:
            return self._build_error_response(req_id, -32600, "Yêu cầu không hợp lệ: thiếu trường method")

        if method == "initialize":
            return self._handle_initialize(req_id)
        elif method == "tools/list":
            return self._handle_tools_list(req_id)
        elif method == "tools/call":
            return self._handle_tools_call(req_id, params)
        elif method == "ping":
            return self._build_success_response(req_id, {"status": "pong"})
        else:
            return self._build_error_response(req_id, -32601, f"Phương thức không hỗ trợ: {method}")

    def _handle_initialize(self, req_id: Any) -> str:
        return self._build_success_response(req_id, {
            "protocolVersion": "2024-11-05",
            "serverInfo": {
                "name": "enterprise-mcp-orchestration-server",
                "version": "2.4.0"
            },
            "capabilities": {
                "tools": {"listChanged": False}
            }
        })

    def _handle_tools_list(self, req_id: Any) -> str:
        tool_list = [asdict(t) for t in self.tools.values()]
        return self._build_success_response(req_id, {"tools": tool_list})

    def _handle_tools_call(self, req_id: Any, params: Dict[str, Any]) -> str:
        name = params.get("name")
        arguments = params.get("arguments", {})

        if name == "inspect_db_schema":
            table = arguments.get("table_name", "")
            return self._tool_inspect_schema(req_id, table)
        elif name == "execute_sandboxed_query":
            query = arguments.get("query", "")
            limit = int(arguments.get("limit", 20))
            return self._tool_execute_query(req_id, query, limit)
        elif name == "git_blame_analyzer":
            file_path = arguments.get("file_path", "")
            start = arguments.get("start_line", 1)
            end = arguments.get("end_line", 50)
            return self._tool_git_blame(req_id, file_path, start, end)
        else:
            return self._build_error_response(req_id, -32602, f"Không tìm thấy công cụ: {name}")

    def _tool_inspect_schema(self, req_id: Any, table_name: str) -> str:
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(f"PRAGMA table_info({table_name});")
                rows = cursor.fetchall()
                if not rows:
                    return self._build_success_response(req_id, {
                        "content": [{"type": "text", "text": f"Bảng '{table_name}' không tồn tại trong cơ sở dữ liệu."}]
                    })
                
                columns = [
                    {"cid": r[0], "name": r[1], "type": r[2], "notnull": bool(r[3]), "dflt_value": r[4], "pk": bool(r[5])}
                    for r in rows
                ]
                return self._build_success_response(req_id, {
                    "content": [{"type": "text", "text": json.dumps({"table": table_name, "columns": columns}, indent=2)}]
                })
        except Exception as e:
            return self._build_error_response(req_id, -32000, f"Kiểm tra schema thất bại: {str(e)}")

    def _tool_execute_query(self, req_id: Any, query: str, limit: int) -> str:
        trimmed = query.strip().upper()
        if not trimmed.startswith("SELECT") and not trimmed.startswith("PRAGMA"):
            return self._build_error_response(req_id, -32000, "Từ chối: Chỉ cho phép câu lệnh SELECT trong sandbox.")

        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(f"{query} LIMIT {min(limit, 100)};")
                col_names = [desc[0] for desc in cursor.description] if cursor.description else []
                rows = cursor.fetchall()
                data = [dict(zip(col_names, row)) for row in rows]
                return self._build_success_response(req_id, {
                    "content": [{"type": "text", "text": json.dumps({"rows_returned": len(data), "data": data}, indent=2)}]
                })
        except Exception as e:
            return self._build_error_response(req_id, -32000, f"Thực thi truy vấn thất bại: {str(e)}")

    def _tool_git_blame(self, req_id: Any, file_path: str, start_line: int, end_line: int) -> str:
        try:
            cmd = ["git", "blame", f"-L{start_line},{end_line}", "--porcelain", file_path]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
            output = result.stdout if result.returncode == 0 else f"Lỗi Git: {result.stderr.strip()}"
            return self._build_success_response(req_id, {
                "content": [{"type": "text", "text": output[:2000]}]
            })
        except Exception as e:
            return self._build_error_response(req_id, -32000, f"Phân tích Git blame thất bại: {str(e)}")

    def _build_success_response(self, req_id: Any, result: Any) -> str:
        return json.dumps({
            "jsonrpc": JSONRPC_VERSION,
            "id": req_id,
            "result": result
        })

    def _build_error_response(self, req_id: Any, code: int, message: str) -> str:
        return json.dumps({
            "jsonrpc": JSONRPC_VERSION,
            "id": req_id,
            "error": {"code": code, "message": message}
        })

    def serve_stdio(self) -> None:
        """Vòng lặp đọc và xử lý yêu cầu qua standard input."""
        for line in sys.stdin:
            line = line.strip()
            if not line:
                continue
            response = self.handle_request(line)
            if response:
                sys.stdout.write(response + "\n")
                sys.stdout.flush()

if __name__ == "__main__":
    server = MCPServer()
    server.serve_stdio()