Agent Protocol

#!/usr/bin/env python3
"""Map and analyze agent capabilities across protocol definitions.

Scans tool schemas, agent cards, and function definitions to produce:
- Capability inventory across agents and protocols
- Conflict detection (duplicate tool names, overlapping capabilities)
- Protocol compatibility matrix
- Gap analysis (missing descriptions, auth, error handling)

Usage:
    python capability_mapper.py *.json
    python capability_mapper.py --scan-dir ./agents/ --json
    python capability_mapper.py agent-card.json mcp-tools.json --detect-conflicts
    python capability_mapper.py --scan-dir ./protocols/ --compatibility-matrix
"""

import argparse
import json
import os
import sys
from collections import defaultdict
from typing import Any

# ---------------------------------------------------------------------------
# Protocol detection (same heuristics as protocol_validator)
# ---------------------------------------------------------------------------

def detect_protocol(data: dict) -> str:
    """Detect which agent protocol a JSON document belongs to."""
    if "inputSchema" in data:
        return "mcp"
    if "skills" in data and ("defaultInputModes" in data or "url" in data):
        return "a2a"
    if "parameters" in data and "name" in data and "inputSchema" not in data:
        return "openai"
    if "tools" in data and isinstance(data["tools"], list):
        sample = data["tools"][0] if data["tools"] else {}
        if "inputSchema" in sample:
            return "mcp"
        if "function" in sample:
            return "openai"
    if "functions" in data:
        return "openai"
    return "unknown"


# ---------------------------------------------------------------------------
# Capability extraction
# ---------------------------------------------------------------------------

def extract_capabilities(filepath: str) -> list[dict]:
    """Extract capability entries from a protocol file.

    Returns a list of capability dicts with normalized fields:
    - source_file, protocol, name, description, parameters (list of param names),
      required_params, has_auth, transport
    """
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (json.JSONDecodeError, OSError) as exc:
        return [{"source_file": filepath, "error": str(exc)}]

    protocol = detect_protocol(data)
    caps: list[dict] = []

    if protocol == "mcp":
        tools = [data] if "name" in data and "inputSchema" in data else data.get("tools", [data])
        for tool in tools:
            if not isinstance(tool, dict) or "name" not in tool:
                continue
            schema = tool.get("inputSchema", {})
            props = list(schema.get("properties", {}).keys())
            required = schema.get("required", [])
            caps.append({
                "source_file": filepath,
                "protocol": "mcp",
                "name": tool.get("name", ""),
                "description": tool.get("description", ""),
                "parameters": props,
                "required_params": required,
                "has_auth": False,
                "transport": "stdio/sse",
            })

    elif protocol == "a2a":
        card_name = data.get("name", "")
        auth = data.get("authentication", {})
        has_auth = bool(auth.get("schemes"))
        caps_section = data.get("capabilities", {})
        streaming = caps_section.get("streaming", False)

        for skill in data.get("skills", []):
            caps.append({
                "source_file": filepath,
                "protocol": "a2a",
                "name": f"{card_name}/{skill.get('id', skill.get('name', ''))}",
                "description": skill.get("description", ""),
                "parameters": [],
                "required_params": [],
                "has_auth": has_auth,
                "transport": "http+json-rpc",
                "tags": skill.get("tags", []),
                "streaming": streaming,
                "agent_name": card_name,
                "agent_url": data.get("url", ""),
            })

        # If no skills declared, still register the agent
        if not data.get("skills"):
            caps.append({
                "source_file": filepath,
                "protocol": "a2a",
                "name": card_name,
                "description": data.get("description", ""),
                "parameters": [],
                "required_params": [],
                "has_auth": has_auth,
                "transport": "http+json-rpc",
                "tags": [],
                "streaming": streaming,
                "agent_name": card_name,
                "agent_url": data.get("url", ""),
            })

    elif protocol == "openai":
        funcs: list[dict] = []
        if "function" in data:
            funcs = [data["function"]]
        elif "functions" in data:
            funcs = data["functions"]
        elif "tools" in data:
            funcs = [t.get("function", t) for t in data["tools"]]
        elif "name" in data:
            funcs = [data]

        for func in funcs:
            params = func.get("parameters", {})
            props = list(params.get("properties", {}).keys())
            required = params.get("required", [])
            caps.append({
                "source_file": filepath,
                "protocol": "openai",
                "name": func.get("name", ""),
                "description": func.get("description", ""),
                "parameters": props,
                "required_params": required,
                "has_auth": False,
                "transport": "http-rest",
                "strict": func.get("strict", False),
            })

    else:
        caps.append({"source_file": filepath, "error": f"Unknown protocol format"})

    return caps


# ---------------------------------------------------------------------------
# Analysis functions
# ---------------------------------------------------------------------------

def detect_conflicts(capabilities: list[dict]) -> list[dict]:
    """Find duplicate or overlapping tool names across files."""
    conflicts: list[dict] = []
    name_map: dict[str, list[dict]] = defaultdict(list)

    for cap in capabilities:
        if "error" in cap:
            continue
        base_name = cap["name"].split("/")[-1]  # strip agent prefix for A2A
        name_map[base_name].append(cap)

    for name, entries in name_map.items():
        if len(entries) > 1:
            sources = [{"file": e["source_file"], "protocol": e["protocol"]} for e in entries]
            # Check if descriptions are semantically similar (simple word overlap)
            descs = [set(e.get("description", "").lower().split()) for e in entries]
            overlap_scores = []
            for i in range(len(descs)):
                for j in range(i + 1, len(descs)):
                    if descs[i] and descs[j]:
                        intersection = descs[i] & descs[j]
                        union = descs[i] | descs[j]
                        score = len(intersection) / len(union) if union else 0
                        overlap_scores.append(score)

            avg_overlap = sum(overlap_scores) / len(overlap_scores) if overlap_scores else 0

            conflicts.append({
                "name": name,
                "count": len(entries),
                "sources": sources,
                "description_similarity": round(avg_overlap, 2),
                "likely_duplicate": avg_overlap > 0.5,
                "protocols": list(set(e["protocol"] for e in entries)),
            })

    return conflicts


def build_compatibility_matrix(capabilities: list[dict]) -> dict:
    """Build a protocol compatibility analysis."""
    protocols: dict[str, list[dict]] = defaultdict(list)
    for cap in capabilities:
        if "error" not in cap:
            protocols[cap["protocol"]].append(cap)

    matrix: dict[str, Any] = {
        "protocols_found": list(protocols.keys()),
        "tool_counts": {p: len(caps) for p, caps in protocols.items()},
        "total_capabilities": sum(len(caps) for caps in protocols.values()),
    }

    # Cross-protocol bridging analysis
    bridge_candidates: list[dict] = []
    all_names: dict[str, dict[str, list[str]]] = defaultdict(lambda: defaultdict(list))
    for cap in capabilities:
        if "error" in cap:
            continue
        base_name = cap["name"].split("/")[-1]
        all_names[base_name][cap["protocol"]].append(cap["source_file"])

    for name, proto_map in all_names.items():
        if len(proto_map) > 1:
            bridge_candidates.append({
                "capability": name,
                "available_in": {p: files for p, files in proto_map.items()},
                "bridge_needed": False,
            })
        elif len(proto_map) == 1:
            proto = list(proto_map.keys())[0]
            missing = [p for p in protocols.keys() if p != proto]
            if missing:
                bridge_candidates.append({
                    "capability": name,
                    "available_in": {proto: proto_map[proto]},
                    "bridge_needed": True,
                    "missing_protocols": missing,
                })

    matrix["bridge_analysis"] = bridge_candidates

    # Feature comparison
    features: dict[str, dict[str, Any]] = {}
    for proto, caps in protocols.items():
        has_auth = any(c.get("has_auth") for c in caps)
        has_streaming = any(c.get("streaming") for c in caps)
        avg_params = sum(len(c.get("parameters", [])) for c in caps) / len(caps) if caps else 0
        desc_coverage = sum(1 for c in caps if c.get("description")) / len(caps) if caps else 0

        features[proto] = {
            "tool_count": len(caps),
            "has_authentication": has_auth,
            "has_streaming": has_streaming,
            "avg_parameters_per_tool": round(avg_params, 1),
            "description_coverage": f"{desc_coverage:.0%}",
        }
    matrix["protocol_features"] = features

    return matrix


def run_gap_analysis(capabilities: list[dict]) -> dict:
    """Identify quality gaps across all capabilities."""
    gaps: dict[str, list[dict]] = {
        "missing_description": [],
        "no_parameters": [],
        "no_required_params": [],
        "short_description": [],
        "no_auth": [],
    }

    for cap in capabilities:
        if "error" in cap:
            continue

        ref = {"name": cap["name"], "file": cap["source_file"], "protocol": cap["protocol"]}

        if not cap.get("description"):
            gaps["missing_description"].append(ref)
        elif len(cap.get("description", "")) < 30:
            gaps["short_description"].append(ref)

        if not cap.get("parameters"):
            gaps["no_parameters"].append(ref)
        elif not cap.get("required_params"):
            gaps["no_required_params"].append(ref)

        if cap["protocol"] in ("a2a",) and not cap.get("has_auth"):
            gaps["no_auth"].append(ref)

    summary: dict[str, Any] = {}
    for gap_type, entries in gaps.items():
        summary[gap_type] = {
            "count": len(entries),
            "items": entries,
        }

    total_caps = sum(1 for c in capabilities if "error" not in c)
    total_issues = sum(len(v) for v in gaps.values())
    summary["quality_score"] = round(
        max(0, (1 - total_issues / max(total_caps * 5, 1))) * 100, 1
    )

    return summary


def build_inventory(capabilities: list[dict]) -> dict:
    """Build a structured inventory of all capabilities."""
    by_protocol: dict[str, list[dict]] = defaultdict(list)
    by_file: dict[str, list[dict]] = defaultdict(list)

    errors: list[dict] = []

    for cap in capabilities:
        if "error" in cap:
            errors.append(cap)
            continue
        entry = {
            "name": cap["name"],
            "description": cap.get("description", "")[:120],
            "protocol": cap["protocol"],
            "parameter_count": len(cap.get("parameters", [])),
            "required_params": len(cap.get("required_params", [])),
            "has_auth": cap.get("has_auth", False),
            "transport": cap.get("transport", "unknown"),
        }
        by_protocol[cap["protocol"]].append(entry)
        by_file[cap["source_file"]].append(entry)

    return {
        "total_capabilities": sum(len(v) for v in by_protocol.values()),
        "by_protocol": dict(by_protocol),
        "by_file": dict(by_file),
        "errors": errors,
    }


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------

def format_human_inventory(inventory: dict) -> str:
    """Format the capability inventory for human reading."""
    lines: list[str] = []
    lines.append("AGENT CAPABILITY INVENTORY")
    lines.append("=" * 60)
    lines.append(f"Total capabilities: {inventory['total_capabilities']}")
    lines.append("")

    for proto, caps in inventory["by_protocol"].items():
        lines.append(f"--- {proto.upper()} ({len(caps)} tool(s)) ---")
        for cap in caps:
            auth_flag = " [AUTH]" if cap["has_auth"] else ""
            lines.append(f"  {cap['name']}{auth_flag}")
            if cap["description"]:
                lines.append(f"    {cap['description']}")
            lines.append(f"    Params: {cap['parameter_count']} ({cap['required_params']} required) | Transport: {cap['transport']}")
        lines.append("")

    if inventory["errors"]:
        lines.append(f"--- ERRORS ({len(inventory['errors'])}) ---")
        for err in inventory["errors"]:
            lines.append(f"  {err.get('source_file', 'unknown')}: {err.get('error', 'unknown error')}")

    return "\n".join(lines)


def format_human_conflicts(conflicts: list[dict]) -> str:
    """Format conflict detection results."""
    lines: list[str] = []
    lines.append("CONFLICT DETECTION REPORT")
    lines.append("=" * 60)

    if not conflicts:
        lines.append("No conflicts detected.")
        return "\n".join(lines)

    lines.append(f"Found {len(conflicts)} potential conflict(s):\n")

    for conf in conflicts:
        dup_flag = " ** LIKELY DUPLICATE **" if conf["likely_duplicate"] else ""
        lines.append(f"  Name: {conf['name']}{dup_flag}")
        lines.append(f"  Occurrences: {conf['count']} across protocols: {', '.join(conf['protocols'])}")
        lines.append(f"  Description similarity: {conf['description_similarity']:.0%}")
        for src in conf["sources"]:
            lines.append(f"    - {src['file']} ({src['protocol']})")
        lines.append("")

    return "\n".join(lines)


def format_human_matrix(matrix: dict) -> str:
    """Format the compatibility matrix."""
    lines: list[str] = []
    lines.append("PROTOCOL COMPATIBILITY MATRIX")
    lines.append("=" * 60)
    lines.append(f"Protocols: {', '.join(matrix['protocols_found'])}")
    lines.append(f"Total capabilities: {matrix['total_capabilities']}")
    lines.append("")

    # Feature comparison table
    features = matrix.get("protocol_features", {})
    if features:
        lines.append("Feature Comparison:")
        lines.append(f"  {'Feature':<30} " + " ".join(f"{p:<12}" for p in features.keys()))
        lines.append("  " + "-" * (30 + 13 * len(features)))

        rows = ["tool_count", "has_authentication", "has_streaming",
                "avg_parameters_per_tool", "description_coverage"]
        for row in rows:
            label = row.replace("_", " ").title()
            vals = []
            for proto in features:
                v = features[proto].get(row, "N/A")
                if isinstance(v, bool):
                    v = "Yes" if v else "No"
                vals.append(str(v))
            lines.append(f"  {label:<30} " + " ".join(f"{v:<12}" for v in vals))
        lines.append("")

    # Bridge analysis
    bridges = matrix.get("bridge_analysis", [])
    needs_bridge = [b for b in bridges if b.get("bridge_needed")]
    if needs_bridge:
        lines.append(f"Bridge Candidates ({len(needs_bridge)} capabilities need cross-protocol bridges):")
        for b in needs_bridge[:20]:  # limit output
            available = ", ".join(b["available_in"].keys())
            missing = ", ".join(b.get("missing_protocols", []))
            lines.append(f"  {b['capability']}: available in [{available}], missing in [{missing}]")
        if len(needs_bridge) > 20:
            lines.append(f"  ... and {len(needs_bridge) - 20} more")
    else:
        lines.append("No protocol bridges needed — all capabilities have cross-protocol coverage.")

    return "\n".join(lines)


def format_human_gaps(gaps: dict) -> str:
    """Format gap analysis results."""
    lines: list[str] = []
    lines.append("GAP ANALYSIS REPORT")
    lines.append("=" * 60)
    lines.append(f"Quality Score: {gaps.get('quality_score', 0)}%\n")

    gap_labels = {
        "missing_description": "Missing Description (critical for LLM tool selection)",
        "short_description": "Short Description (< 30 chars)",
        "no_parameters": "No Parameters Defined",
        "no_required_params": "No Required Parameters",
        "no_auth": "No Authentication (A2A agents)",
    }

    for gap_key, label in gap_labels.items():
        info = gaps.get(gap_key, {"count": 0, "items": []})
        count = info["count"]
        status = "PASS" if count == 0 else "FAIL"
        lines.append(f"  [{status}] {label}: {count}")
        if count > 0:
            for item in info["items"][:5]:
                lines.append(f"        - {item['name']} ({item['protocol']}) in {item['file']}")
            if count > 5:
                lines.append(f"        ... and {count - 5} more")
        lines.append("")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# File discovery
# ---------------------------------------------------------------------------

def find_json_files(directory: str) -> list[str]:
    """Recursively find JSON files in a directory."""
    files: list[str] = []
    for root, _dirs, filenames in os.walk(directory):
        for fn in sorted(filenames):
            if fn.endswith(".json"):
                files.append(os.path.join(root, fn))
    return files


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main() -> int:
    parser = argparse.ArgumentParser(
        description="Map and analyze agent capabilities across protocol definitions.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  %(prog)s tools.json agent-card.json\n"
            "  %(prog)s --scan-dir ./agents/\n"
            "  %(prog)s --detect-conflicts mcp-tools.json openai-funcs.json\n"
            "  %(prog)s --compatibility-matrix --scan-dir ./protocols/ --json\n"
            "  %(prog)s --gap-analysis *.json\n"
        ),
    )
    parser.add_argument("files", nargs="*", metavar="FILE",
                        help="JSON protocol files to analyze")
    parser.add_argument("--scan-dir", metavar="DIR",
                        help="Recursively scan directory for JSON protocol files")
    parser.add_argument("--detect-conflicts", action="store_true",
                        help="Detect duplicate or overlapping tool names")
    parser.add_argument("--compatibility-matrix", action="store_true",
                        help="Generate cross-protocol compatibility matrix")
    parser.add_argument("--gap-analysis", action="store_true",
                        help="Run quality gap analysis across all capabilities")
    parser.add_argument("--json", dest="json_output", action="store_true",
                        help="Output results as JSON")

    args = parser.parse_args()

    # Collect files
    all_files: list[str] = list(args.files)
    if args.scan_dir:
        if not os.path.isdir(args.scan_dir):
            print(f"Error: Directory not found: {args.scan_dir}", file=sys.stderr)
            return 1
        all_files.extend(find_json_files(args.scan_dir))

    if not all_files:
        parser.print_help()
        print("\nError: No input files specified. Provide files or use --scan-dir.", file=sys.stderr)
        return 1

    # Extract capabilities from all files
    all_capabilities: list[dict] = []
    for filepath in all_files:
        if not os.path.isfile(filepath):
            all_capabilities.append({"source_file": filepath, "error": "File not found"})
            continue
        all_capabilities.extend(extract_capabilities(filepath))

    valid_caps = [c for c in all_capabilities if "error" not in c]

    if not valid_caps and not any("error" in c for c in all_capabilities):
        print("No capabilities found in the provided files.", file=sys.stderr)
        return 1

    # Determine which analyses to run
    run_conflicts = args.detect_conflicts
    run_matrix = args.compatibility_matrix
    run_gaps = args.gap_analysis

    # If no specific analysis requested, run inventory + all analyses
    if not (run_conflicts or run_matrix or run_gaps):
        run_conflicts = True
        run_matrix = True
        run_gaps = True

    # Build results
    results: dict[str, Any] = {}
    results["inventory"] = build_inventory(all_capabilities)

    if run_conflicts:
        results["conflicts"] = detect_conflicts(valid_caps)
    if run_matrix:
        results["compatibility_matrix"] = build_compatibility_matrix(valid_caps)
    if run_gaps:
        results["gap_analysis"] = run_gap_analysis(valid_caps)

    # Output
    if args.json_output:
        print(json.dumps(results, indent=2))
    else:
        print(format_human_inventory(results["inventory"]))
        if run_conflicts:
            print("\n")
            print(format_human_conflicts(results["conflicts"]))
        if run_matrix:
            print("\n")
            print(format_human_matrix(results["compatibility_matrix"]))
        if run_gaps:
            print("\n")
            print(format_human_gaps(results["gap_analysis"]))

    # Exit code: 1 if there are errors or likely duplicates
    has_errors = bool(results["inventory"].get("errors"))
    has_dupes = any(c.get("likely_duplicate") for c in results.get("conflicts", []))
    return 1 if (has_errors or has_dupes) else 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Validate agent protocol configurations for MCP, A2A, and OpenAI Function Calling compliance.

Reads JSON tool schema files and checks them against protocol-specific rules:
- MCP: tool name conventions, description quality, inputSchema structure, required fields
- A2A: agent card completeness, skill definitions, capability declarations
- OpenAI: function calling schema, parameter types, strict mode readiness

Usage:
    python protocol_validator.py schema.json
    python protocol_validator.py --protocol mcp tools/*.json
    python protocol_validator.py --protocol a2a agent-card.json --json
    python protocol_validator.py --strict schema.json
"""

import argparse
import json
import os
import re
import sys
from typing import Any

# ---------------------------------------------------------------------------
# Severity levels
# ---------------------------------------------------------------------------
SEVERITY_ERROR = "error"
SEVERITY_WARNING = "warning"
SEVERITY_INFO = "info"


def _issue(severity: str, code: str, message: str, path: str = "") -> dict:
    entry = {"severity": severity, "code": code, "message": message}
    if path:
        entry["path"] = path
    return entry


# ---------------------------------------------------------------------------
# JSON Schema type helpers
# ---------------------------------------------------------------------------
VALID_JSON_TYPES = {"string", "number", "integer", "boolean", "array", "object", "null"}
VALID_STRING_FORMATS = {
    "date", "date-time", "time", "email", "uri", "uri-reference",
    "hostname", "ipv4", "ipv6", "uuid", "regex",
}


def _validate_json_schema_fragment(schema: dict, path: str) -> list[dict]:
    """Validate a JSON Schema fragment for structural correctness."""
    issues: list[dict] = []
    if not isinstance(schema, dict):
        issues.append(_issue(SEVERITY_ERROR, "SCHEMA_NOT_OBJECT",
                             f"Schema at '{path}' must be an object, got {type(schema).__name__}", path))
        return issues

    stype = schema.get("type")
    if stype and stype not in VALID_JSON_TYPES:
        issues.append(_issue(SEVERITY_ERROR, "INVALID_TYPE",
                             f"Unknown JSON Schema type '{stype}'", path))

    if stype == "array":
        items = schema.get("items")
        if items is None:
            issues.append(_issue(SEVERITY_WARNING, "ARRAY_NO_ITEMS",
                                 "Array type should declare 'items' schema", path))
        elif isinstance(items, dict):
            issues.extend(_validate_json_schema_fragment(items, f"{path}.items"))

    if stype == "object":
        props = schema.get("properties", {})
        for pname, pdef in props.items():
            issues.extend(_validate_json_schema_fragment(pdef, f"{path}.properties.{pname}"))

    fmt = schema.get("format")
    if fmt and stype == "string" and fmt not in VALID_STRING_FORMATS:
        issues.append(_issue(SEVERITY_INFO, "UNKNOWN_FORMAT",
                             f"Non-standard string format '{fmt}' — agents may ignore it", path))

    if "enum" in schema:
        if not isinstance(schema["enum"], list) or len(schema["enum"]) == 0:
            issues.append(_issue(SEVERITY_ERROR, "EMPTY_ENUM",
                                 "Enum must be a non-empty array", path))

    return issues


# ---------------------------------------------------------------------------
# MCP validation
# ---------------------------------------------------------------------------
MCP_NAME_RE = re.compile(r"^[a-z][a-z0-9]*(_[a-z0-9]+)*$")
DESCRIPTION_TRIGGER_PHRASES = {"use when", "use this", "returns", "use for"}


def validate_mcp_tool(tool: dict, strict: bool = False) -> list[dict]:
    """Validate a single MCP tool definition."""
    issues: list[dict] = []

    # --- name ---
    name = tool.get("name")
    if not name:
        issues.append(_issue(SEVERITY_ERROR, "MISSING_NAME", "Tool definition missing 'name' field"))
    elif not MCP_NAME_RE.match(name):
        issues.append(_issue(SEVERITY_ERROR, "BAD_NAME_FORMAT",
                             f"Tool name '{name}' must be snake_case (lowercase alphanumeric + underscores)"))
    elif "_" not in name:
        issues.append(_issue(SEVERITY_WARNING, "NAME_NO_VERB_NOUN",
                             f"Tool name '{name}' should follow verb_noun pattern (e.g., search_documents)"))

    # --- description ---
    desc = tool.get("description", "")
    if not desc:
        issues.append(_issue(SEVERITY_ERROR, "MISSING_DESCRIPTION",
                             "Tool must have a description for LLM tool selection"))
    else:
        if len(desc) < 30:
            issues.append(_issue(SEVERITY_WARNING, "SHORT_DESCRIPTION",
                                 f"Description is only {len(desc)} chars — aim for 50+ for effective LLM guidance"))
        if len(desc) > 1024:
            issues.append(_issue(SEVERITY_WARNING, "LONG_DESCRIPTION",
                                 f"Description is {len(desc)} chars — consider shortening to under 1024"))
        desc_lower = desc.lower()
        if not any(phrase in desc_lower for phrase in DESCRIPTION_TRIGGER_PHRASES):
            issues.append(_issue(SEVERITY_WARNING, "DESCRIPTION_NO_USAGE_GUIDANCE",
                                 "Description should include usage guidance (e.g., 'Use when...', 'Returns...')"))
        if strict and not desc.rstrip().endswith("."):
            issues.append(_issue(SEVERITY_INFO, "DESCRIPTION_NO_PERIOD",
                                 "Description should end with a period for consistency"))

    # --- inputSchema ---
    schema = tool.get("inputSchema")
    if schema is None:
        issues.append(_issue(SEVERITY_WARNING, "MISSING_INPUT_SCHEMA",
                             "Tool has no inputSchema — consider adding one even if no parameters are needed"))
    elif not isinstance(schema, dict):
        issues.append(_issue(SEVERITY_ERROR, "INPUT_SCHEMA_NOT_OBJECT",
                             f"inputSchema must be an object, got {type(schema).__name__}"))
    else:
        if schema.get("type") != "object":
            issues.append(_issue(SEVERITY_ERROR, "INPUT_SCHEMA_TYPE",
                                 "inputSchema.type must be 'object'"))

        props = schema.get("properties", {})
        required = schema.get("required", [])

        if not isinstance(required, list):
            issues.append(_issue(SEVERITY_ERROR, "REQUIRED_NOT_ARRAY",
                                 "inputSchema.required must be an array"))
        else:
            for req_name in required:
                if req_name not in props:
                    issues.append(_issue(SEVERITY_ERROR, "REQUIRED_MISSING_PROP",
                                         f"Required field '{req_name}' not found in properties"))

        for pname, pdef in props.items():
            ppath = f"inputSchema.properties.{pname}"
            if not pdef.get("description"):
                issues.append(_issue(SEVERITY_WARNING, "PROP_NO_DESCRIPTION",
                                     f"Property '{pname}' missing description — LLMs need context", ppath))
            if not pdef.get("type"):
                issues.append(_issue(SEVERITY_WARNING, "PROP_NO_TYPE",
                                     f"Property '{pname}' missing type declaration", ppath))
            issues.extend(_validate_json_schema_fragment(pdef, ppath))

        if strict and not props:
            issues.append(_issue(SEVERITY_INFO, "NO_PROPERTIES",
                                 "inputSchema has no properties — is this intentional?"))

    return issues


# ---------------------------------------------------------------------------
# A2A Agent Card validation
# ---------------------------------------------------------------------------
def validate_a2a_agent_card(card: dict, strict: bool = False) -> list[dict]:
    """Validate a Google A2A agent card."""
    issues: list[dict] = []

    for field in ("name", "description", "url", "version"):
        if not card.get(field):
            issues.append(_issue(SEVERITY_ERROR, f"MISSING_{field.upper()}",
                                 f"Agent card missing required field '{field}'"))

    url = card.get("url", "")
    if url and not url.startswith(("http://", "https://")):
        issues.append(_issue(SEVERITY_ERROR, "INVALID_URL",
                             f"Agent URL '{url}' must start with http:// or https://"))

    provider = card.get("provider", {})
    if not provider.get("organization"):
        issues.append(_issue(SEVERITY_WARNING, "NO_PROVIDER_ORG",
                             "Agent card should include provider.organization"))

    caps = card.get("capabilities", {})
    if not isinstance(caps, dict):
        issues.append(_issue(SEVERITY_ERROR, "CAPABILITIES_NOT_OBJECT",
                             "capabilities must be an object"))
    else:
        for cap_key in ("streaming", "pushNotifications", "stateTransitionHistory"):
            if cap_key not in caps:
                issues.append(_issue(SEVERITY_INFO, f"NO_{cap_key.upper()}",
                                     f"capabilities.{cap_key} not declared — defaults to false"))

    auth = card.get("authentication", {})
    if not auth:
        issues.append(_issue(SEVERITY_WARNING, "NO_AUTH",
                             "Agent card has no authentication section"))
    else:
        if not auth.get("schemes"):
            issues.append(_issue(SEVERITY_WARNING, "NO_AUTH_SCHEMES",
                                 "authentication.schemes should list supported auth mechanisms"))

    skills = card.get("skills", [])
    if not skills:
        issues.append(_issue(SEVERITY_WARNING, "NO_SKILLS",
                             "Agent card should declare at least one skill"))
    for idx, skill in enumerate(skills):
        spath = f"skills[{idx}]"
        for sf in ("id", "name", "description"):
            if not skill.get(sf):
                issues.append(_issue(SEVERITY_ERROR, f"SKILL_MISSING_{sf.upper()}",
                                     f"Skill at {spath} missing '{sf}'", spath))
        if strict and not skill.get("examples"):
            issues.append(_issue(SEVERITY_INFO, "SKILL_NO_EXAMPLES",
                                 f"Skill at {spath} has no examples — adding examples improves discoverability",
                                 spath))

    input_modes = card.get("defaultInputModes", [])
    output_modes = card.get("defaultOutputModes", [])
    if not input_modes:
        issues.append(_issue(SEVERITY_INFO, "NO_INPUT_MODES",
                             "No defaultInputModes declared"))
    if not output_modes:
        issues.append(_issue(SEVERITY_INFO, "NO_OUTPUT_MODES",
                             "No defaultOutputModes declared"))

    return issues


# ---------------------------------------------------------------------------
# OpenAI Function Calling validation
# ---------------------------------------------------------------------------
def validate_openai_function(func: dict, strict: bool = False) -> list[dict]:
    """Validate an OpenAI function calling definition."""
    issues: list[dict] = []

    name = func.get("name")
    if not name:
        issues.append(_issue(SEVERITY_ERROR, "MISSING_NAME",
                             "Function definition missing 'name' field"))
    elif not re.match(r"^[a-zA-Z_][a-zA-Z0-9_-]*$", name):
        issues.append(_issue(SEVERITY_ERROR, "BAD_FUNCTION_NAME",
                             f"Function name '{name}' contains invalid characters"))
    if name and len(name) > 64:
        issues.append(_issue(SEVERITY_ERROR, "NAME_TOO_LONG",
                             f"Function name '{name}' exceeds 64-character limit"))

    desc = func.get("description", "")
    if not desc:
        issues.append(_issue(SEVERITY_ERROR, "MISSING_DESCRIPTION",
                             "Function must have a description"))
    elif len(desc) > 1024:
        issues.append(_issue(SEVERITY_WARNING, "LONG_DESCRIPTION",
                             f"Description is {len(desc)} chars — OpenAI recommends under 1024"))

    params = func.get("parameters")
    if params is not None:
        if params.get("type") != "object":
            issues.append(_issue(SEVERITY_ERROR, "PARAMS_TYPE",
                                 "parameters.type must be 'object'"))
        props = params.get("properties", {})
        for pname, pdef in props.items():
            ppath = f"parameters.properties.{pname}"
            issues.extend(_validate_json_schema_fragment(pdef, ppath))
            if not pdef.get("description"):
                issues.append(_issue(SEVERITY_WARNING, "PROP_NO_DESCRIPTION",
                                     f"Property '{pname}' missing description", ppath))

        if strict:
            required = params.get("required", [])
            additional = params.get("additionalProperties")
            if additional is not False:
                issues.append(_issue(SEVERITY_INFO, "STRICT_ADDITIONAL_PROPS",
                                     "For OpenAI strict mode, set additionalProperties: false"))
            for pname in props:
                if pname not in required:
                    issues.append(_issue(SEVERITY_INFO, "STRICT_ALL_REQUIRED",
                                         f"Strict mode requires all properties in 'required' — '{pname}' is optional"))

    return issues


# ---------------------------------------------------------------------------
# Auto-detect protocol
# ---------------------------------------------------------------------------
def detect_protocol(data: dict) -> str:
    """Heuristically detect which protocol a schema belongs to."""
    if "inputSchema" in data:
        return "mcp"
    if "skills" in data or "defaultInputModes" in data or "defaultOutputModes" in data:
        return "a2a"
    if "parameters" in data and "name" in data:
        return "openai"
    # Check if it wraps a list of tools
    if "tools" in data and isinstance(data["tools"], list):
        first = data["tools"][0] if data["tools"] else {}
        if "inputSchema" in first:
            return "mcp"
        if "function" in first:
            return "openai"
    return "unknown"


def validate_file(filepath: str, protocol: str, strict: bool) -> dict:
    """Validate a single file and return results."""
    result: dict[str, Any] = {"file": filepath, "protocol": protocol, "issues": []}

    try:
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
    except json.JSONDecodeError as exc:
        result["issues"].append(_issue(SEVERITY_ERROR, "INVALID_JSON",
                                       f"Failed to parse JSON: {exc}"))
        return result
    except OSError as exc:
        result["issues"].append(_issue(SEVERITY_ERROR, "FILE_READ_ERROR", str(exc)))
        return result

    detected = protocol if protocol != "auto" else detect_protocol(data)
    result["protocol"] = detected

    if detected == "mcp":
        # Could be a single tool or a list
        tools = [data] if "name" in data else data.get("tools", [data])
        for idx, tool in enumerate(tools):
            tool_issues = validate_mcp_tool(tool, strict)
            for iss in tool_issues:
                iss["tool_index"] = idx
                iss.setdefault("path", "")
            result["issues"].extend(tool_issues)

    elif detected == "a2a":
        result["issues"].extend(validate_a2a_agent_card(data, strict))

    elif detected == "openai":
        funcs = [data] if "name" in data else [f.get("function", f) for f in data.get("tools", data.get("functions", [data]))]
        for idx, func in enumerate(funcs):
            func_issues = validate_openai_function(func, strict)
            for iss in func_issues:
                iss["function_index"] = idx
            result["issues"].extend(func_issues)

    else:
        result["issues"].append(_issue(SEVERITY_ERROR, "UNKNOWN_PROTOCOL",
                                       "Could not detect protocol — use --protocol to specify"))

    return result


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------
SEVERITY_SYMBOLS = {
    SEVERITY_ERROR: "[ERROR]  ",
    SEVERITY_WARNING: "[WARN]   ",
    SEVERITY_INFO: "[INFO]   ",
}


def format_human(results: list[dict]) -> str:
    """Format validation results for human consumption."""
    lines: list[str] = []
    total_errors = 0
    total_warnings = 0

    for res in results:
        lines.append(f"\n{'='*60}")
        lines.append(f"File:     {res['file']}")
        lines.append(f"Protocol: {res['protocol']}")

        if not res["issues"]:
            lines.append("Status:   PASS — no issues found")
            continue

        errors = [i for i in res["issues"] if i["severity"] == SEVERITY_ERROR]
        warnings = [i for i in res["issues"] if i["severity"] == SEVERITY_WARNING]
        infos = [i for i in res["issues"] if i["severity"] == SEVERITY_INFO]
        total_errors += len(errors)
        total_warnings += len(warnings)

        lines.append(f"Status:   {len(errors)} error(s), {len(warnings)} warning(s), {len(infos)} info(s)")
        lines.append("-" * 60)

        for iss in res["issues"]:
            prefix = SEVERITY_SYMBOLS.get(iss["severity"], "         ")
            loc = ""
            if "tool_index" in iss:
                loc = f"[tool {iss['tool_index']}] "
            elif "function_index" in iss:
                loc = f"[func {iss['function_index']}] "
            if iss.get("path"):
                loc += f"({iss['path']}) "
            lines.append(f"  {prefix}{loc}{iss['code']}: {iss['message']}")

    lines.append(f"\n{'='*60}")
    lines.append(f"Total: {len(results)} file(s), {total_errors} error(s), {total_warnings} warning(s)")

    if total_errors > 0:
        lines.append("Result: FAIL")
    elif total_warnings > 0:
        lines.append("Result: PASS with warnings")
    else:
        lines.append("Result: PASS")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> int:
    parser = argparse.ArgumentParser(
        description="Validate agent protocol configurations (MCP, A2A, OpenAI Function Calling).",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  %(prog)s tool.json\n"
            "  %(prog)s --protocol mcp tools/*.json\n"
            "  %(prog)s --protocol a2a agent-card.json --json\n"
            "  %(prog)s --strict --protocol openai functions.json\n"
        ),
    )
    parser.add_argument("files", nargs="+", metavar="FILE",
                        help="JSON schema file(s) to validate")
    parser.add_argument("--protocol", choices=["mcp", "a2a", "openai", "auto"],
                        default="auto",
                        help="Protocol to validate against (default: auto-detect)")
    parser.add_argument("--strict", action="store_true",
                        help="Enable strict validation with additional checks")
    parser.add_argument("--json", dest="json_output", action="store_true",
                        help="Output results as JSON")
    args = parser.parse_args()

    results: list[dict] = []
    for filepath in args.files:
        if not os.path.isfile(filepath):
            results.append({
                "file": filepath,
                "protocol": args.protocol,
                "issues": [_issue(SEVERITY_ERROR, "FILE_NOT_FOUND",
                                  f"File not found: {filepath}")]
            })
            continue
        results.append(validate_file(filepath, args.protocol, args.strict))

    if args.json_output:
        print(json.dumps({"results": results, "file_count": len(results)}, indent=2))
    else:
        print(format_human(results))

    has_errors = any(
        iss["severity"] == SEVERITY_ERROR
        for res in results
        for iss in res["issues"]
    )
    return 1 if has_errors else 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Generate tool schema definitions for MCP, A2A, and OpenAI Function Calling.

Creates well-structured protocol schemas from:
- Inline parameter definitions (--param name:type:description)
- JSON config files describing tools
- Python function signatures (parsed from source files)

Usage:
    python schema_generator.py --name search_documents --desc "Search docs" --param query:string:required:"Search query"
    python schema_generator.py --from-config tools.json --protocol mcp
    python schema_generator.py --from-python module.py --function my_func --protocol openai
    python schema_generator.py --name research --protocol a2a --skill-tags research,web --json
"""

import argparse
import ast
import json
import os
import re
import sys
from typing import Any, Optional

# ---------------------------------------------------------------------------
# Python type -> JSON Schema type mapping
# ---------------------------------------------------------------------------
PY_TYPE_MAP: dict[str, dict[str, Any]] = {
    "str": {"type": "string"},
    "int": {"type": "integer"},
    "float": {"type": "number"},
    "bool": {"type": "boolean"},
    "list": {"type": "array"},
    "dict": {"type": "object"},
    "List": {"type": "array"},
    "Dict": {"type": "object"},
    "Optional": {},  # handled separately
    "None": {"type": "null"},
    "NoneType": {"type": "null"},
}

CLI_TYPE_MAP: dict[str, str] = {
    "string": "string",
    "str": "string",
    "int": "integer",
    "integer": "integer",
    "float": "number",
    "number": "number",
    "bool": "boolean",
    "boolean": "boolean",
    "array": "array",
    "list": "array",
    "object": "object",
    "dict": "object",
}


# ---------------------------------------------------------------------------
# Parameter parsing from CLI --param flags
# ---------------------------------------------------------------------------
def parse_param_spec(spec: str) -> dict:
    """Parse a parameter specification string.

    Format: name:type[:required|optional][:description]
    Examples:
        query:string:required:"The search query"
        limit:integer:optional:"Max results"
        tags:array:"List of tags"
    """
    # Handle quoted descriptions
    parts: list[str] = []
    current = ""
    in_quotes = False
    for ch in spec:
        if ch == '"' or ch == "'":
            in_quotes = not in_quotes
        elif ch == ':' and not in_quotes:
            parts.append(current)
            current = ""
            continue
        current += ch
    parts.append(current)

    if len(parts) < 2:
        raise ValueError(f"Parameter spec needs at least name:type — got '{spec}'")

    name = parts[0].strip()
    ptype = CLI_TYPE_MAP.get(parts[1].strip().lower(), "string")

    required = True
    description = ""

    for part in parts[2:]:
        stripped = part.strip().strip('"').strip("'")
        if stripped.lower() == "required":
            required = True
        elif stripped.lower() == "optional":
            required = False
        else:
            description = stripped

    result: dict[str, Any] = {"name": name, "type": ptype, "required": required}
    if description:
        result["description"] = description
    return result


# ---------------------------------------------------------------------------
# Python source parsing
# ---------------------------------------------------------------------------
def extract_function_schema(source_path: str, function_name: str) -> dict:
    """Extract parameter schema from a Python function's signature and docstring."""
    with open(source_path, "r", encoding="utf-8") as f:
        tree = ast.parse(f.read(), filename=source_path)

    func_node: Optional[ast.FunctionDef] = None
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if node.name == function_name:
                func_node = node
                break

    if func_node is None:
        raise ValueError(f"Function '{function_name}' not found in {source_path}")

    # Extract docstring
    docstring = ast.get_docstring(func_node) or ""

    # Parse docstring for parameter descriptions (Google/NumPy style)
    param_docs: dict[str, str] = {}
    doc_lines = docstring.split("\n")
    in_params = False
    current_param = ""
    for line in doc_lines:
        stripped = line.strip()
        if stripped.lower() in ("args:", "parameters:", "params:"):
            in_params = True
            continue
        if in_params:
            if stripped.lower() in ("returns:", "raises:", "yields:", "examples:", "note:", "notes:"):
                in_params = False
                continue
            # Match "param_name (type): description" or "param_name: description"
            m = re.match(r"(\w+)\s*(?:\([^)]*\))?\s*:\s*(.*)", stripped)
            if m:
                current_param = m.group(1)
                param_docs[current_param] = m.group(2).strip()
            elif current_param and stripped:
                param_docs[current_param] += " " + stripped

    # Extract parameters from function signature
    params: list[dict] = []
    args = func_node.args

    # Count defaults to determine which args are optional
    num_defaults = len(args.defaults)
    num_args = len(args.args)

    for idx, arg in enumerate(args.args):
        if arg.arg in ("self", "cls"):
            continue

        param: dict[str, Any] = {"name": arg.arg}

        # Resolve type annotation
        if arg.annotation:
            type_str = _annotation_to_string(arg.annotation)
            is_optional, base_type = _parse_type_string(type_str)
            json_type = PY_TYPE_MAP.get(base_type, {}).get("type", "string")
            param["type"] = json_type
            param["required"] = not is_optional
        else:
            param["type"] = "string"
            param["required"] = True

        # Check if it has a default value
        default_idx = idx - (num_args - num_defaults)
        if default_idx >= 0:
            param["required"] = False
            default_node = args.defaults[default_idx]
            default_val = _extract_default(default_node)
            if default_val is not None:
                param["default"] = default_val

        if arg.arg in param_docs:
            param["description"] = param_docs[arg.arg]

        params.append(param)

    # First line of docstring as function description
    func_desc = doc_lines[0].strip() if doc_lines else ""

    return {
        "name": function_name,
        "description": func_desc,
        "params": params,
    }


def _annotation_to_string(node: ast.expr) -> str:
    """Convert an AST annotation node to a string representation."""
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Constant):
        return str(node.value)
    if isinstance(node, ast.Attribute):
        return f"{_annotation_to_string(node.value)}.{node.attr}"
    if isinstance(node, ast.Subscript):
        base = _annotation_to_string(node.value)
        sl = _annotation_to_string(node.slice)
        return f"{base}[{sl}]"
    if isinstance(node, ast.Tuple):
        return ", ".join(_annotation_to_string(e) for e in node.elts)
    if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr):
        left = _annotation_to_string(node.left)
        right = _annotation_to_string(node.right)
        return f"{left} | {right}"
    return "Any"


def _parse_type_string(type_str: str) -> tuple[bool, str]:
    """Parse a type string and return (is_optional, base_type)."""
    is_optional = False
    # Handle Optional[X] and X | None
    m = re.match(r"Optional\[(\w+)]", type_str)
    if m:
        return True, m.group(1)
    if "| None" in type_str or "None |" in type_str:
        base = type_str.replace("| None", "").replace("None |", "").strip()
        return True, base
    return is_optional, type_str.split("[")[0]


def _extract_default(node: ast.expr) -> Any:
    """Extract a default value from an AST node."""
    if isinstance(node, ast.Constant):
        return node.value
    if isinstance(node, ast.List):
        return []
    if isinstance(node, ast.Dict):
        return {}
    if isinstance(node, ast.Name) and node.id == "None":
        return None
    return None


# ---------------------------------------------------------------------------
# Schema generation for each protocol
# ---------------------------------------------------------------------------
def generate_mcp_schema(name: str, description: str, params: list[dict]) -> dict:
    """Generate an MCP tool schema."""
    properties: dict[str, Any] = {}
    required: list[str] = []

    for p in params:
        prop: dict[str, Any] = {"type": p["type"]}
        if p.get("description"):
            prop["description"] = p["description"]
        if "default" in p:
            prop["default"] = p["default"]
        if "enum" in p:
            prop["enum"] = p["enum"]
        properties[p["name"]] = prop
        if p.get("required", True):
            required.append(p["name"])

    schema: dict[str, Any] = {
        "name": name,
        "description": description,
        "inputSchema": {
            "type": "object",
            "properties": properties,
        },
    }
    if required:
        schema["inputSchema"]["required"] = required

    return schema


def generate_openai_schema(name: str, description: str, params: list[dict],
                           strict: bool = False) -> dict:
    """Generate an OpenAI function calling schema."""
    properties: dict[str, Any] = {}
    required: list[str] = []

    for p in params:
        prop: dict[str, Any] = {"type": p["type"]}
        if p.get("description"):
            prop["description"] = p["description"]
        if "enum" in p:
            prop["enum"] = p["enum"]
        properties[p["name"]] = prop
        if p.get("required", True) or strict:
            required.append(p["name"])

    func_def: dict[str, Any] = {
        "name": name,
        "description": description,
        "parameters": {
            "type": "object",
            "properties": properties,
        },
    }
    if required:
        func_def["parameters"]["required"] = required
    if strict:
        func_def["parameters"]["additionalProperties"] = False
        func_def["strict"] = True

    return {"type": "function", "function": func_def}


def generate_a2a_agent_card(name: str, description: str, url: str = "https://agent.example.com",
                            org: str = "Organization", skill_tags: list[str] | None = None,
                            streaming: bool = False) -> dict:
    """Generate an A2A agent card."""
    card: dict[str, Any] = {
        "name": name,
        "description": description,
        "url": url,
        "provider": {
            "organization": org,
            "url": f"https://{org.lower().replace(' ', '-')}.example.com",
        },
        "version": "1.0.0",
        "capabilities": {
            "streaming": streaming,
            "pushNotifications": False,
            "stateTransitionHistory": True,
        },
        "authentication": {
            "schemes": ["Bearer"],
            "credentials": "oauth2",
        },
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/plain", "application/json"],
        "skills": [
            {
                "id": name.lower().replace(" ", "-"),
                "name": name,
                "description": description,
                "tags": skill_tags or [],
                "examples": [],
            }
        ],
    }
    return card


# ---------------------------------------------------------------------------
# Config file parsing
# ---------------------------------------------------------------------------
def load_config(filepath: str) -> list[dict]:
    """Load tool definitions from a JSON config file.

    Expected format:
    {
      "tools": [
        {
          "name": "tool_name",
          "description": "...",
          "params": [
            {"name": "query", "type": "string", "required": true, "description": "..."}
          ]
        }
      ]
    }
    """
    with open(filepath, "r", encoding="utf-8") as f:
        data = json.load(f)

    if isinstance(data, list):
        return data
    if "tools" in data:
        return data["tools"]
    # Single tool
    return [data]


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------
def format_human_schema(schema: dict, protocol: str) -> str:
    """Format a schema for human-readable output."""
    lines: list[str] = []
    lines.append(f"Protocol: {protocol.upper()}")
    lines.append(f"{'='*50}")

    if protocol == "mcp":
        lines.append(f"Tool:        {schema.get('name', 'N/A')}")
        lines.append(f"Description: {schema.get('description', 'N/A')}")
        isc = schema.get("inputSchema", {})
        props = isc.get("properties", {})
        required = set(isc.get("required", []))
        if props:
            lines.append(f"\nParameters ({len(props)}):")
            for pname, pdef in props.items():
                req_marker = "*" if pname in required else " "
                ptype = pdef.get("type", "any")
                pdesc = pdef.get("description", "")
                default = f" [default: {pdef['default']}]" if "default" in pdef else ""
                lines.append(f"  {req_marker} {pname}: {ptype}{default}")
                if pdesc:
                    lines.append(f"      {pdesc}")
            lines.append("\n  * = required")

    elif protocol == "openai":
        func = schema.get("function", schema)
        lines.append(f"Function:    {func.get('name', 'N/A')}")
        lines.append(f"Description: {func.get('description', 'N/A')}")
        params = func.get("parameters", {})
        props = params.get("properties", {})
        required = set(params.get("required", []))
        if props:
            lines.append(f"\nParameters ({len(props)}):")
            for pname, pdef in props.items():
                req_marker = "*" if pname in required else " "
                ptype = pdef.get("type", "any")
                pdesc = pdef.get("description", "")
                lines.append(f"  {req_marker} {pname}: {ptype}")
                if pdesc:
                    lines.append(f"      {pdesc}")
            lines.append("\n  * = required")
        if func.get("strict"):
            lines.append("\n  [STRICT MODE enabled]")

    elif protocol == "a2a":
        lines.append(f"Agent:       {schema.get('name', 'N/A')}")
        lines.append(f"Description: {schema.get('description', 'N/A')}")
        lines.append(f"URL:         {schema.get('url', 'N/A')}")
        lines.append(f"Version:     {schema.get('version', 'N/A')}")
        caps = schema.get("capabilities", {})
        lines.append(f"\nCapabilities:")
        for k, v in caps.items():
            lines.append(f"  {k}: {v}")
        skills = schema.get("skills", [])
        if skills:
            lines.append(f"\nSkills ({len(skills)}):")
            for s in skills:
                lines.append(f"  - {s.get('name', 'N/A')}: {s.get('description', '')}")
                if s.get("tags"):
                    lines.append(f"    Tags: {', '.join(s['tags'])}")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> int:
    parser = argparse.ArgumentParser(
        description="Generate tool schema definitions for MCP, A2A, and OpenAI protocols.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            '  %(prog)s --name search_docs --desc "Search documents." '
            '--param query:string:required:"Search query" --param limit:int:optional:"Max results"\n'
            "  %(prog)s --from-python api.py --function search_docs --protocol openai\n"
            "  %(prog)s --from-config tools.json --protocol mcp --json\n"
            '  %(prog)s --name "Research Agent" --protocol a2a --skill-tags research,web\n'
        ),
    )

    source_group = parser.add_argument_group("input source (pick one)")
    source_group.add_argument("--name", help="Tool/agent name")
    source_group.add_argument("--desc", "--description", dest="description", default="",
                              help="Tool/agent description")
    source_group.add_argument("--param", action="append", dest="params", metavar="SPEC",
                              help="Parameter spec: name:type[:required|optional][:description]")
    source_group.add_argument("--from-python", metavar="FILE",
                              help="Extract schema from a Python function signature")
    source_group.add_argument("--function", help="Function name to extract (with --from-python)")
    source_group.add_argument("--from-config", metavar="FILE",
                              help="Load tool definitions from a JSON config file")

    output_group = parser.add_argument_group("output options")
    output_group.add_argument("--protocol", choices=["mcp", "a2a", "openai"], default="mcp",
                              help="Target protocol (default: mcp)")
    output_group.add_argument("--strict", action="store_true",
                              help="Enable strict mode (OpenAI: all required + no additionalProperties)")
    output_group.add_argument("--json", dest="json_output", action="store_true",
                              help="Output as JSON (default for piping)")
    output_group.add_argument("--output", "-o", metavar="FILE",
                              help="Write output to file instead of stdout")

    # A2A-specific options
    a2a_group = parser.add_argument_group("A2A options")
    a2a_group.add_argument("--url", default="https://agent.example.com",
                           help="Agent URL for A2A agent card")
    a2a_group.add_argument("--org", default="Organization",
                           help="Provider organization name")
    a2a_group.add_argument("--skill-tags", default="",
                           help="Comma-separated skill tags for A2A agent card")
    a2a_group.add_argument("--streaming", action="store_true",
                           help="Enable streaming capability in A2A agent card")

    args = parser.parse_args()

    schemas: list[dict] = []

    # Source: config file
    if args.from_config:
        if not os.path.isfile(args.from_config):
            print(f"Error: Config file not found: {args.from_config}", file=sys.stderr)
            return 1
        tools = load_config(args.from_config)
        for tool in tools:
            name = tool.get("name", "unnamed_tool")
            desc = tool.get("description", "")
            params = tool.get("params", [])
            if args.protocol == "mcp":
                schemas.append(generate_mcp_schema(name, desc, params))
            elif args.protocol == "openai":
                schemas.append(generate_openai_schema(name, desc, params, args.strict))
            elif args.protocol == "a2a":
                tags = [t.strip() for t in args.skill_tags.split(",") if t.strip()]
                schemas.append(generate_a2a_agent_card(name, desc, args.url, args.org, tags, args.streaming))

    # Source: Python function
    elif args.from_python:
        if not os.path.isfile(args.from_python):
            print(f"Error: Python file not found: {args.from_python}", file=sys.stderr)
            return 1
        if not args.function:
            print("Error: --function is required with --from-python", file=sys.stderr)
            return 1
        try:
            extracted = extract_function_schema(args.from_python, args.function)
        except ValueError as exc:
            print(f"Error: {exc}", file=sys.stderr)
            return 1
        name = extracted["name"]
        desc = extracted["description"]
        params = extracted["params"]
        if args.protocol == "mcp":
            schemas.append(generate_mcp_schema(name, desc, params))
        elif args.protocol == "openai":
            schemas.append(generate_openai_schema(name, desc, params, args.strict))
        elif args.protocol == "a2a":
            tags = [t.strip() for t in args.skill_tags.split(",") if t.strip()]
            schemas.append(generate_a2a_agent_card(name, desc, args.url, args.org, tags, args.streaming))

    # Source: inline params
    elif args.name:
        params = []
        for spec in (args.params or []):
            try:
                params.append(parse_param_spec(spec))
            except ValueError as exc:
                print(f"Error parsing param: {exc}", file=sys.stderr)
                return 1

        if args.protocol == "mcp":
            schemas.append(generate_mcp_schema(args.name, args.description, params))
        elif args.protocol == "openai":
            schemas.append(generate_openai_schema(args.name, args.description, params, args.strict))
        elif args.protocol == "a2a":
            tags = [t.strip() for t in args.skill_tags.split(",") if t.strip()]
            schemas.append(generate_a2a_agent_card(
                args.name, args.description, args.url, args.org, tags, args.streaming))
    else:
        parser.print_help()
        return 1

    # Output
    if len(schemas) == 1:
        output_data = schemas[0]
    else:
        output_data = {"tools": schemas}

    if args.json_output or args.output:
        output_text = json.dumps(output_data, indent=2)
    else:
        parts = []
        for s in schemas:
            parts.append(format_human_schema(s, args.protocol))
        output_text = "\n\n".join(parts)
        output_text += "\n\n--- JSON ---\n" + json.dumps(output_data, indent=2)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output_text + "\n")
        print(f"Schema written to {args.output}")
    else:
        print(output_text)

    return 0


if __name__ == "__main__":
    sys.exit(main())