flowchart TD
    subgraph Agent Framework ["LLM Orchestrator / Agent"]
        A[Agent Context Call] -->|MCP Tool Request| B[MCPSkillRuntime Entrypoint]
    end

    subgraph Runtime Control Plane ["mcp-skill-runtime Engine"]
        B --> C{MCP Schema Parser & Validator}
        C -->|Schema Invalid| D[Structured Validation Error Return]
        C -->|Schema Valid| E[Token-Bucket Rate Limiter]
        
        E -->|Bucket Exhausted| F[Rate Limit Exceeded Exception]
        E -->|Token Granted| G[Async Isolation Sandbox]
        
        subgraph Execution Sandbox ["Deterministic Sandbox"]
            G --> H[Execution Worker]
            H --> I{Timeout Guard}
            I -->|Exceeds Timeout Limit| J[Kill Task & Trigger Retry Isolation]
            J -->|Max Retries Reached| K[Timeout Error Payload]
            I -->|Execution Success| L[Standardized Tool Payload]
        end
    end

    subgraph Observability ["Telemetry & Monitoring Cockpit"]
        D --> M[(SQLite Trace Store)]
        F --> M
        K --> M
        L --> M
        M --> N[Real-Time Telemetry Cockpit Dashboard]
    end

    L -->|Strict Schema Result| A
    D -->|Error Payload| A
    F -->|Throttled Payload| A
    K -->|Timeout Payload| A

# Using uv (Recommended for performance)
uv pip install mcp-skill-runtime

# Using standard pip
pip install mcp-skill-runtime

import asyncio
from typing import Dict, Any
from pydantic import BaseModel, Field
from mcp_skill_runtime import (
    SkillEngine,
    BaseSkill,
    SkillConfig,
    RateLimitConfig,
    ExecutionResult
)

# 1. Define input parameters using Pydantic schema validation
class DatabaseQueryInput(BaseModel):
    query: str = Field(..., description="SQL SELECT query string to execute.")
    max_rows: int = Field(default=100, ge=1, le=1000, description="Row return limit.")

# 2. Implement the Skill harness logic
class DatabaseQuerySkill(BaseSkill):
    name = "db_query_tool"
    description = "Safely executes read-only queries against the analytics store."
    input_schema = DatabaseQueryInput

    async def execute(self, params: DatabaseQueryInput) -> Dict[str, Any]:
        # Simulate business logic / external call
        await asyncio.sleep(0.05)
        return {
            "status": "SUCCESS",
            "rows_returned": 2,
            "data": [
                {"id": 101, "metric": "cpu_utilization", "value": 42.1},
                {"id": 102, "metric": "memory_utilization", "value": 68.4}
            ]
        }

async def main():
    # 3. Initialize the Engine with telemetry storage
    engine = SkillEngine(db_path="mcp_telemetry.db")

    # 4. Register skill with precise rate limiting and timeout boundaries
    engine.register_skill(
        skill=DatabaseQuerySkill(),
        config=SkillConfig(
            timeout_seconds=2.0,
            max_retries=2,
            rate_limit=RateLimitConfig(
                tokens_per_minute=60,
                burst_capacity=5
            )
        )
    )

    # 5. Invocation payload (Valid Call)
    print("--- Invoking Valid Payload ---")
    result: ExecutionResult = await engine.invoke(
        skill_name="db_query_tool",
        payload={"query": "SELECT * FROM metrics", "max_rows": 50}
    )

    print(f"Success: {result.success}")
    print(f"Data: {result.data}")
    print(f"Execution Latency: {result.metrics.latency_ms:.2f}ms")

    # 6. Invocation payload (Invalid Schema Call)
    print("\n--- Invoking Malformed Payload ---")
    invalid_result: ExecutionResult = await engine.invoke(
        skill_name="db_query_tool",
        payload={"query": "SELECT * FROM metrics", "max_rows": 99999} # Exceeds le=1000 limit
    )

    print(f"Success: {invalid_result.success}")
    print(f"Error Code: {invalid_result.error.code}")
    print(f"Validation Details: {invalid_result.error.message}")

if __name__ == "__main__":
    asyncio.run(main())

# Verify installation & print spec metadata
mcp-runtime info

# Start the MCP Tool Execution Server (JSON-RPC 2.0 / MCP Protocol)
mcp-runtime serve --port 8080 --host 127.0.0.1 --db-path ./mcp_telemetry.db

# Invoke a specific registered skill directly via CLI
mcp-runtime invoke \
  --skill db_query_tool \
  --payload '{"query": "SELECT 1", "max_rows": 10}' \
  --db-path ./mcp_telemetry.db

# Launch the Visual Telemetry Cockpit Dashboard
mcp-runtime cockpit --port 3000 --db-path ./mcp_telemetry.db