API: Core

from mcp_eval.core import (  task, # Define a test task  with_agent, # Specify which agent to use  parametrize, # Run tests with multiple inputs  setup, # Run before tests  teardown, # Run after tests  TestResult # Test execution results ) 

from mcp_eval.core import task from mcp_eval.catalog import Expect  @task("My first test") async def test_basic(agent, session):  """A simple test that checks basic functionality."""  response = await agent.generate_str("Hello, world!")    await session.assert_that(  Expect.content.contains("Hello"),  response=response  ) 

# Compose multiple assertions and give them names for clear reporting @task("Fetch and summarize") async def test_fetch_and_summarize(agent, session):  response = await agent.generate_str(  "Fetch https://example.com and summarize in one sentence"  )   await session.assert_that(  Expect.tools.was_called("fetch"),  name="fetch_called"  )  await session.assert_that(  Expect.content.contains("Example Domain"),  response=response,  name="has_expected_text"  )  await session.assert_that(  Expect.performance.max_iterations(3),  name="efficient"  ) 

@task("Expensive test") async def test_with_config(agent, session):  response = await agent.generate_str("Complex analysis task")  # Test continues... 

@with_agent("default") @task("Test with default agent") async def test_default(agent):  # Uses the 'default' agent from config  response = await agent.generate_str("Test prompt") 

# Define agents in mcpeval.yaml first @with_agent("specialized_agent") @task("Test specialized behavior") async def test_specialized(agent):  # Uses a different agent configuration  pass  @with_agent("minimal_agent") @task("Test minimal setup") async def test_minimal(agent):  # Uses yet another configuration  pass 

from mcp_agent.agents.agent_spec import AgentSpec  @with_agent(AgentSpec(  name="code_reviewer",  instruction="You are a code reviewer. Be thorough and critical.",  server_names=["filesystem", "git"] )) @task("Code review test") async def test_code_review(agent):  response = await agent.generate_str("Review this code: ...") 

@with_agent("default") @parametrize("number", [1, 2, 5, 10, 100]) @task("Test with different numbers") async def test_numbers(agent, number):  response = await agent.generate_str(f"Is {number} prime?")  # Each number creates a separate test case  await agent.assert_that(  Expect.content.regex(r"(yes|no|prime|composite)")  ) 

@with_agent("default") @parametrize("operation", ["add", "subtract", "multiply"]) @parametrize("x", [1, 10, 100]) @parametrize("y", [2, 5]) @task("Test calculator operations") async def test_calculator(agent, operation, x, y):  # Creates 3 * 3 * 2 = 18 test cases!  prompt = f"Use the calculator to {operation} {x} and {y}"  response = await agent.generate_str(prompt)    # Verify the right tool was used  await agent.assert_that(  Expect.tools.was_called(f"calculator_{operation}")  ) 

@with_agent("default") @parametrize(  "name,url,expected",  [  ("home", "https://example.com", "Example Domain"),  ("httpbin_json", "https://httpbin.org/json", "slideshow"),  ], ) @task("Fetch {name}") async def test_fetch_case(agent, session, name: str, url: str, expected: str):  response = await agent.generate_str(f"Fetch {url}")   await session.assert_that(  Expect.tools.was_called("fetch"),  name=f"{name}_fetch_called",  )  await session.assert_that(  Expect.content.contains(expected, case_sensitive=False),  response=response,  name=f"{name}_has_expected",  ) 

def get_test_cases():  """Generate test cases dynamically."""  import json  with open("test_data.json") as f:  return json.load(f)["test_cases"]  @with_agent("default") @parametrize("test_case", get_test_cases()) @task("Dynamic test cases") async def test_dynamic(agent, test_case):  response = await agent.generate_str(test_case["prompt"])    await agent.assert_that(  Expect.content.contains(test_case["expected"]),  response=response  ) 

from mcp_eval.core import setup, teardown import os import tempfile  test_dir = None  @setup def prepare_test_environment():  """Create temporary test directory."""  global test_dir  test_dir = tempfile.mkdtemp(prefix="mcp_test_")  print(f"🚀 Created test directory: {test_dir}")    # Set up test files  with open(f"{test_dir}/test.txt", "w") as f:  f.write("Test content")  @teardown def cleanup_test_environment():  """Clean up after tests."""  global test_dir  if test_dir and os.path.exists(test_dir):  import shutil  shutil.rmtree(test_dir)  print(f"🧹 Cleaned up {test_dir}") 

@setup async def async_prepare():  """Setup that requires async operations."""  # Connect to database  await db.connect()    # Seed test data  await db.execute("INSERT INTO test_table ...")    print("✅ Database ready")  @teardown async def async_cleanup():  """Async cleanup operations."""  await db.execute("DELETE FROM test_table WHERE ...")  await db.disconnect() 

@setup def validate_environment():  """Ensure test environment is properly configured."""  import sys    # Check Python version  if sys.version_info < (3, 10):  raise RuntimeError("Tests require Python 3.10+")    # Check required environment variables  required_vars = ["ANTHROPIC_API_KEY", "TEST_SERVER_URL"]  missing = [var for var in required_vars if not os.getenv(var)]    if missing:  raise RuntimeError(f"Missing environment variables: {missing}")    # Check MCP servers are accessible  from mcp_eval.utils import check_server_health  if not check_server_health("my_server"):  raise RuntimeError("MCP server 'my_server' is not responding")    print("✅ Environment validated successfully") 

from mcp_eval.core import TestResult  # After a test runs, you get a TestResult: result = TestResult(  id="test_123_abc",  name="Test basic fetch",  passed=True,  duration_ms=1234.56,  parameters={"url": "https://example.com"},  metrics={  "tool_calls": 2,  "tokens_used": 500,  "cost_usd": 0.01  },  evaluations=[  {"name": "content_check", "passed": True, "score": 1.0},  {"name": "performance", "passed": True, "details": "Under 2s"}  ],  error=None # Or error message if failed ) 

# Use CLI combined reports or session.get_metrics()/get_results() for summaries 

def analyze_test_results(results: list[TestResult]):  """Analyze a batch of test results."""  total = len(results)  passed = sum(1 for r in results if r.passed)    total_duration = sum(r.duration_ms for r in results)  total_cost = sum(r.metrics.get('cost_usd', 0) for r in results)    print(f"\n📊 Test Summary:")  print(f" Total tests: {total}")  print(f" Passed: {passed}/{total} ({passed/total*100:.1f}%)")  print(f" Total duration: {total_duration/1000:.2f}s")  print(f" Total cost: ${total_cost:.4f}")    # Find slowest tests  slowest = sorted(results, key=lambda r: r.duration_ms, reverse=True)[:3]  print(f"\n🐢 Slowest tests:")  for result in slowest:  print(f" {result.name}: {result.duration_ms:.0f}ms")    # Find failed tests  failed = [r for r in results if not r.passed]  if failed:  print(f"\n❌ Failed tests:")  for result in failed:  print(f" {result.name}: {result.error}") 

# Run a single test function (pytest-style selector supported by the runner for decorator tests) mcp-eval run tests/test_fetch.py::test_fetch_case  # Run pytest tests (use pytest) uv run pytest -q tests 

import os, pytest from mcp_eval.core import task  @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skip on CI") @task("Local-only behavior") async def test_local_only(agent, session):  response = await agent.generate_str("Do something local")  # assertions...  @pytest.mark.slow @task("Slow end-to-end scenario") async def test_slow_scenario(agent, session):  # long-running flow...  ... 

pytest -m "not slow" tests/ 

from contextvars import ContextVar  test_context = ContextVar('test_context', default={})  @task("Test with context") async def test_with_context(agent, session):  # Set context for this test  ctx = test_context.get().copy()  ctx['test_id'] = session.test_id  ctx['start_time'] = time.time()  test_context.set(ctx)    response = await agent.generate_str("Test prompt")    # Context is available throughout the test  duration = time.time() - ctx['start_time']  print(f"Test {ctx['test_id']} took {duration:.2f}s") 

@with_agent("default") @task("Test error recovery") async def test_error_handling(agent):  # Trigger an error condition  response = await agent.generate_str("Divide 10 by 0")    # Verify graceful handling  await agent.assert_that(  Expect.content.regex(r"(error|cannot|undefined|infinity)"),  name="handles_division_by_zero"  )    # Verify no tool crashes  # Check tool success via success_rate, e.g., Expect.tools.success_rate(1.0) 

@with_agent("default") @task("Test complete workflow") async def test_workflow(agent, session):  # Step 1: Authentication  auth_response = await agent.generate_str("Authenticate as test_user")  await session.assert_that(  Expect.tools.was_called("auth"),  name="authentication_attempted"  )    # Step 2: Fetch data  data_response = await agent.generate_str("Get my profile data")  await session.assert_that(  Expect.tools.was_called("fetch_profile"),  name="profile_fetched"  )    # Step 3: Process  process_response = await agent.generate_str("Summarize my activity")  await session.assert_that(  Expect.content.contains("summary"),  response=process_response,  name="summary_generated"  )    # Verify the complete sequence  await session.assert_that(  Expect.tools.sequence(["auth", "fetch_profile", "summarize"]),  name="correct_workflow_order"  )