mcp-tester 0.5.1

Comprehensive MCP server testing tool - library and CLI
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

//! MCP Protocol Conformance Test Suite
//!
//! Validates any MCP server against the MCP protocol spec (2025-11-25).
//! Scenarios are grouped by domain: Core, Tools, Resources, Prompts, Tasks.
//! Each domain reports independently -- a server with no resources passes
//! if it correctly reports empty capabilities.

pub(crate) mod core_domain;
pub(crate) mod prompts;
pub(crate) mod resources;
pub(crate) mod tasks;
pub(crate) mod tools;

use crate::report::{TestCategory, TestReport, TestResult};
use crate::tester::ServerTester;
use pmcp::types::ServerCapabilities;
use std::time::Instant;

/// MCP protocol domain for conformance filtering.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConformanceDomain {
    Core,
    Tools,
    Resources,
    Prompts,
    Tasks,
}

impl ConformanceDomain {
    /// Parse domain name from string (case-insensitive).
    pub fn from_str_loose(s: &str) -> Option<Self> {
        match s.to_lowercase().as_str() {
            "core" => Some(Self::Core),
            "tools" => Some(Self::Tools),
            "resources" => Some(Self::Resources),
            "prompts" => Some(Self::Prompts),
            "tasks" => Some(Self::Tasks),
            _ => None,
        }
    }
}

/// Check if a server advertises a capability. Returns `Some(skip_result)` if
/// the capability is absent, `None` if present.
pub(crate) fn check_capability(
    tester: &ServerTester,
    domain_name: &str,
    category: TestCategory,
    has_capability: impl FnOnce(&ServerCapabilities) -> bool,
) -> Option<Vec<TestResult>> {
    let has = tester.server_capabilities().is_some_and(has_capability);
    if !has {
        Some(vec![TestResult::skipped(
            format!("{domain_name}: capability not advertised"),
            category,
            format!("Server does not advertise {domain_name} capability"),
        )])
    } else {
        None
    }
}

/// Orchestrates conformance test execution across domains.
pub struct ConformanceRunner {
    strict: bool,
    domains: Option<Vec<ConformanceDomain>>,
}

impl ConformanceRunner {
    /// Create a new runner. If `domains` is None, all domains run.
    pub fn new(strict: bool, domains: Option<Vec<ConformanceDomain>>) -> Self {
        Self { strict, domains }
    }

    /// Run conformance tests against the server.
    /// Core domain always runs first (handles initialization).
    /// Other domains are skipped if core fails or if domain is filtered out.
    pub async fn run(&self, tester: &mut ServerTester) -> TestReport {
        let mut report = TestReport::new();
        let start = Instant::now();

        // Core always runs first -- it initializes the server connection
        if self.should_run(ConformanceDomain::Core) {
            for result in core_domain::run_core_conformance(tester).await {
                report.add_test(result);
            }
        }

        // Only proceed with other domains if core didn't fail
        if !report.has_failures() {
            if self.should_run(ConformanceDomain::Tools) {
                for result in tools::run_tools_conformance(tester).await {
                    report.add_test(result);
                }
            }

            if self.should_run(ConformanceDomain::Resources) {
                for result in resources::run_resources_conformance(tester).await {
                    report.add_test(result);
                }
            }

            if self.should_run(ConformanceDomain::Prompts) {
                for result in prompts::run_prompts_conformance(tester).await {
                    report.add_test(result);
                }
            }

            if self.should_run(ConformanceDomain::Tasks) {
                for result in tasks::run_tasks_conformance(tester).await {
                    report.add_test(result);
                }
            }
        }

        if self.strict {
            report.apply_strict_mode();
        }

        report.duration = start.elapsed();
        report
    }

    fn should_run(&self, domain: ConformanceDomain) -> bool {
        self.domains.as_ref().is_none_or(|d| d.contains(&domain))
    }
}