Architecture-as-Code/ikt-arh-kultura-valodu-tehnologijas/scripts/analyze_mcpf_response.py

#!/usr/bin/env python3
"""
MCPF Initialize Response Analyzer
Tests MCPF dual-layer trust integration and displays the response in a structured format.
"""

import json
import requests
import sys
from typing import Dict, Any

def test_mcpf_initialize(endpoint: str = "http://localhost:8787/mcp") -> Dict[str, Any]:
    """
    Send initialize request to MCP server and return parsed response.
    """
    payload = {
        "jsonrpc": "2.0",
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-03-26",
            "capabilities": {},
            "clientInfo": {
                "name": "mcpf-analyzer",
                "version": "1.0"
            }
        },
        "id": 1
    }
    
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream"
    }
    
    print(f"🔍 Testing endpoint: {endpoint}")
    print(f"📤 Sending initialize request...\n")
    
    try:
        response = requests.post(endpoint, json=payload, headers=headers, timeout=5)
        response.raise_for_status()
        
        # Parse SSE response (format: "event: message\ndata: {json}\n")
        lines = response.text.strip().split('\n')
        data_line = None
        for line in lines:
            if line.startswith('data: '):
                data_line = line[6:]  # Remove "data: " prefix
                break
        
        if not data_line:
            print("❌ No data line found in SSE response")
            return {}
        
        return json.loads(data_line)
    
    except requests.exceptions.RequestException as e:
        print(f"❌ Request failed: {e}")
        return {}
    except json.JSONDecodeError as e:
        print(f"❌ JSON parsing failed: {e}")
        return {}

def print_section(title: str, content: str):
    """Print a formatted section."""
    print("=" * 80)
    print(f"  {title}")
    print("=" * 80)
    print(content)
    print()

def analyze_response(response: Dict[str, Any]):
    """Analyze and display MCPF response structure."""
    
    if not response:
        print("❌ No response received")
        return
    
    # Check for result
    result = response.get("result", {})
    if not result:
        print("❌ No result in response")
        print_section("Full Response", json.dumps(response, indent=2))
        return
    
    print("✅ Initialize successful\n")
    
    # Display basic server info
    server_info = result.get("serverInfo", {})
    print_section(
        "Server Information",
        f"Name: {server_info.get('name', 'Unknown')}\n"
        f"Version: {server_info.get('version', 'Unknown')}\n"
        f"Protocol: {result.get('protocolVersion', 'Unknown')}"
    )
    
    # Display _meta (MCPF Layer 1)
    meta = result.get("_meta")
    if not meta:
        print("⚠️  No _meta field found - MCPF integration not present")
        return
    
    print("✅ MCPF Layer 1 (Session-Level Trust Metadata) present\n")
    
    # Identity section
    identity = meta.get("identity", {})
    if identity:
        print_section(
            "🆔 Identity (DID & Keys)",
            f"DID: {identity.get('id', 'Not found')}\n"
            f"MCP Service: {identity.get('service', {}).get('mcp', 'Not found')}\n"
            f"JWKS URI: {identity.get('keys', {}).get('jwks_uri', 'Not found')}"
        )
    
    # MCPF section
    mcpf = meta.get("mcpf", {})
    if mcpf:
        print_section(
            "📋 MCPF Metadata",
            f"Version: {mcpf.get('version', 'Not found')}\n"
            f"Spec Repository: {mcpf.get('spec', {}).get('repository', 'Not found')}\n"
            f"Manifest URL: {mcpf.get('entrypoint', {}).get('url', 'Not found')}\n"
            f"Trust Registry: {mcpf.get('artifacts', {}).get('trust_registry', 'Not found')}\n"
            f"Credential: {mcpf.get('artifacts', {}).get('credential', 'Not found')}"
        )
    
    # Trust verifications section
    trust = meta.get("trust", {})
    verifications = trust.get("verifications", [])
    if verifications:
        print_section("🔐 Trust Verifications", "")
        for i, verification in enumerate(verifications, 1):
            print(f"  Verification #{i}:")
            print(f"    Verifier: {verification.get('verifier', 'Not found')}")
            print(f"    Type: {', '.join(verification.get('type', []))}")
            print(f"    Covers: {verification.get('covers', 'Not found')}")
            print(f"    Credential: {verification.get('credential', 'Not found')}")
            
            proof = verification.get('proof_hint', {})
            if proof:
                print(f"    Proof:")
                print(f"      Method: {proof.get('verificationMethod', 'Not found')}")
                print(f"      Created: {proof.get('created', 'Not found')}")
            print()
    
    # Summary
    print_section(
        "✅ MCPF Integration Summary",
        f"• Identity DID present: {'✅' if identity else '❌'}\n"
        f"• MCPF metadata present: {'✅' if mcpf else '❌'}\n"
        f"• Trust verifications: {len(verifications)}\n"
        f"• VeriTrust verified: {'✅' if any(v.get('verifier') == 'did:web:veritrust.vc' for v in verifications) else '❌'}\n"
        f"\n🎯 This server implements MCPF v{mcpf.get('version', '?')} dual-layer trust"
    )
    
    # Full JSON for reference
    print_section("📄 Complete _meta JSON", json.dumps(meta, indent=2))

def main():
    endpoint = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8787/mcp"
    
    print("""
╔════════════════════════════════════════════════════════════════════════════╗
║                 MCPF Initialize Response Analyzer                          ║
║                   Testing Dual-Layer Trust Integration                     ║
╚════════════════════════════════════════════════════════════════════════════╝
""")
    
    response = test_mcpf_initialize(endpoint)
    analyze_response(response)
    
    print("""
╔════════════════════════════════════════════════════════════════════════════╗
║  Next: Test Layer 2 (per-response attestations) by calling a tool         ║
║  Example: {"method": "tools/call", "params": {"name": "search", ...}}     ║
╚════════════════════════════════════════════════════════════════════════════╝
""")

if __name__ == "__main__":
    main()