JEM Trajectory Evaluation for MCP Tool Calls

┌────────────────────────────────────────────────────────┐
│               AGENT TRAJECTORY TRACE                   │
├───────────────┬────────────────────────┬───────────────┤
│ Step 1        │ Step 2                 │ Step 3 (Final)│
│ MCP Tool:     │ MCP Tool:              │ Assertion:    │
│ tools/list    │ tools/call (postgres)  │ State Verify  │
├───────────────┴────────────────────────┴───────────────┤
│                JEM EVALUATOR ENGINE                    │
│   [Schema Check] -> [Step Logic] -> [Semantic Judge]   │
└────────────────────────────────────────────────────────┘

# jem_mcp_evaluator.py
import json
from typing import List, Dict, Any
from pydantic import BaseModel, Field

class ToolExecutionStep(BaseModel):
    step_index: int
    tool_name: str
    arguments: Dict[str, Any]
    output: Any
    latency_ms: int

class JEMScore(BaseModel):
    trajectory_id: str
    step_validity_score: float = Field(ge=0.0, le=1.0)
    outcome_accuracy_score: float = Field(ge=0.0, le=1.0)
    passed: bool
    failure_step: int | None = None
    diagnostics: str

class MCPEvaluator:
    def __init__(self, judge_client, schemas: Dict[str, Any]):
        self.client = judge_client
        self.schemas = schemas

    def evaluate_trajectory(
        self,
        trajectory_id: str,
        goal: str,
        steps: List[ToolExecutionStep],
        expected_state: Dict[str, Any]
    ) -> JEMScore:
        # Phase 1: Deterministic Schema & Type Verification
        for step in steps:
            schema = self.schemas.get(step.tool_name)
            if not schema:
                return JEMScore(
                    trajectory_id=trajectory_id,
                    step_validity_score=0.0,
                    outcome_accuracy_score=0.0,
                    passed=False,
                    failure_step=step.step_index,
                    diagnostics=f"Unknown tool invoked: {step.tool_name}"
                )

        # Phase 2: Semantic Trajectory Validation via Judge Model
        judge_prompt = f"""
You are a JEM (Judged Exact Match) evaluator for Model Context Protocol agents.
User Goal: {goal}
Execution Steps Taken:
{json.dumps([s.model_dump() for s in steps], indent=2)}

Expected Environment State:
{json.dumps(expected_state, indent=2)}

Score the trajectory on:
1. Step Efficiency: No redundant or circular tool calls.
2. Parameter Accuracy: Correct filter keys, valid SQL, safe queries.
3. Outcome Equivalence: Did the tools produce the expected result?

Respond with JSON:
{{
  "step_validity": 0.0 to 1.0,
  "outcome_accuracy": 0.0 to 1.0,
  "reasoning": "string"
}}
"""
        # Call judge model (e.g., Claude 3.7 Sonnet or Gemini 2.5 Flash)
        judge_res = self.client.complete(judge_prompt)
        parsed = json.loads(judge_res.text)

        step_score = parsed["step_validity"]
        outcome_score = parsed["outcome_accuracy"]
        passed = (step_score >= 0.8) and (outcome_score >= 0.9)

        return JEMScore(
            trajectory_id=trajectory_id,
            step_validity_score=step_score,
            outcome_accuracy_score=outcome_score,
            passed=passed,
            diagnostics=parsed.get("reasoning", "")
        )

# .github/workflows/mcp-agent-eval.yml
name: MCP Agent JEM Regression Suite
on: [push, pull_request]

jobs:
  evaluate-mcp-trajectories:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Run JEM Benchmarks
        run: |
          pip install pydantic pytest
          python -m pytest tests/eval/test_mcp_trajectories.py --junitxml=results.xml