Part 6: From Coder to Orchestrator — Multi-Agent Swarms, Model Context Protocol (MCP 2.0) & Workflow Systems

package main

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

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

// AgentRole defines specialized functional domains within the swarm.
type AgentRole string

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

// SwarmTask encapsulates an atomic unit of work assigned to a sub-agent.
type SwarmTask struct {
	ID        string    `json:"id"`
	Target    AgentRole `json:"target"`
	Payload   string    `json:"payload"`
	Timeout   time.Duration `json:"timeout"`
}

// AgentHandoffArtifact represents the immutable output delivered by a sub-agent.
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 manages concurrent sub-agent execution pools and synchronization.
type SwarmOrchestrator struct {
	mu        sync.Mutex
	artifacts []AgentHandoffArtifact
}

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

// DispatchSwarm executes tasks concurrently with bounded worker contexts.
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 failed on task %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("Swarm execution failed: %v", err)
	}

	fmt.Printf("=== Swarm Completed Successfully (%d Artifacts) ===\n", len(artifacts))
	for _, art := range artifacts {
		fmt.Printf("[%s] Role: %-15s -> Artifact: %s\n", art.Status, art.Agent, art.Result)
	}
}

#!/usr/bin/env python3
"""
Production Model Context Protocol (MCP 2.0) Server.
Implements JSON-RPC 2.0 protocol over standard input/output streams.
Provides schema inspection, sandboxed query execution, and git blame telemetry.
"""

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

# Protocol Constants
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="Inspect table schemas, columns, constraints, and data types in the database.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "table_name": {"type": "string", "description": "Name of the table to inspect"}
                    },
                    "required": ["table_name"]
                }
            ),
            "execute_sandboxed_query": ToolDefinition(
                name="execute_sandboxed_query",
                description="Execute a safe, read-only SQL SELECT query with a row limit.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "SQL SELECT query to execute"},
                        "limit": {"type": "integer", "description": "Maximum rows to return", "default": 20}
                    },
                    "required": ["query"]
                }
            ),
            "git_blame_analyzer": ToolDefinition(
                name="git_blame_analyzer",
                description="Analyze git commit author, timestamp, and recency for a file range.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "file_path": {"type": "string", "description": "Repository file path to inspect"},
                        "start_line": {"type": "integer", "description": "Start line number"},
                        "end_line": {"type": "integer", "description": "End line number"}
                    },
                    "required": ["file_path"]
                }
            )
        }

    def _init_sample_db(self) -> None:
        """Initialize in-memory database with test tables and sample data."""
        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]:
        """Process an incoming JSON-RPC 2.0 request."""
        try:
            req = json.loads(request_str)
        except json.JSONDecodeError as e:
            return self._build_error_response(None, -32700, f"Parse error: {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, "Invalid Request: missing method")

        # Dispatch methods
        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"Method not found: {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"Unknown tool: {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"Table '{table_name}' does not exist."}]
                    })
                
                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"Schema inspection failed: {str(e)}")

    def _tool_execute_query(self, req_id: Any, query: str, limit: int) -> str:
        # Enforce read-only constraint
        trimmed = query.strip().upper()
        if not trimmed.startswith("SELECT") and not trimmed.startswith("PRAGMA"):
            return self._build_error_response(req_id, -32000, "Forbidden: Only SELECT queries are permitted in 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"Query execution failed: {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"Git error: {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"Git blame execution failed: {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:
        """Run standard I/O server loop."""
        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()
    # In interactive test or stdio pipeline
    server.serve_stdio()