|
| 1 | +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. |
| 2 | +# |
| 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +# you may not use this file except in compliance with the License. |
| 5 | +# You may obtain a copy of the License at |
| 6 | +# |
| 7 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +# |
| 9 | +# Unless required by applicable law or agreed to in writing, software |
| 10 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +# See the License for the specific language governing permissions and |
| 13 | +# limitations under the License. |
| 14 | + |
| 15 | +"""Utility for validating MCP client examples. |
| 16 | +
|
| 17 | +Extracts framework classes and method calls from example code, |
| 18 | +then validates they can be imported in the framework's environment. |
| 19 | +""" |
| 20 | + |
| 21 | +import ast |
| 22 | +import pytest |
| 23 | +import subprocess |
| 24 | +from pathlib import Path |
| 25 | + |
| 26 | + |
| 27 | +class ExampleValidator: |
| 28 | + """Validates MCP client examples by extracting and testing API usage.""" |
| 29 | + |
| 30 | + # Skip standard library and utility modules to focus on framework APIs |
| 31 | + IGNORED_MODULES = {'asyncio', 'dotenv', 'os', 'warnings'} |
| 32 | + |
| 33 | + def extract_imports_and_classes( |
| 34 | + self, main_file: Path |
| 35 | + ) -> tuple[list[str], set[str], dict[str, set[str]]]: |
| 36 | + """Extract framework imports, classes, and method calls from example file. |
| 37 | +
|
| 38 | + Returns: |
| 39 | + imports: List of import statements for validation script |
| 40 | + classes: Set of class names to test for __init__ method |
| 41 | + method_calls: Dict mapping object names to their called methods |
| 42 | + """ |
| 43 | + tree = ast.parse(main_file.read_text(encoding='utf-8')) |
| 44 | + imports, classes, method_calls = [], set(), {} |
| 45 | + ignored_names = set() # Track names from ignored modules |
| 46 | + |
| 47 | + # Walk AST to find imports and method calls |
| 48 | + for node in ast.walk(tree): |
| 49 | + if isinstance(node, ast.Import): |
| 50 | + self._process_import(node, imports, ignored_names) |
| 51 | + elif isinstance(node, ast.ImportFrom): |
| 52 | + self._process_import_from(node, imports, classes, ignored_names) |
| 53 | + elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): |
| 54 | + self._process_method_call(node, method_calls, ignored_names) |
| 55 | + |
| 56 | + return imports, classes, method_calls |
| 57 | + |
| 58 | + def _process_import( |
| 59 | + self, node: ast.Import, imports: list[str], ignored_names: set[str] |
| 60 | + ) -> None: |
| 61 | + """Process 'import' statements.""" |
| 62 | + for alias in node.names: |
| 63 | + if alias.name in self.IGNORED_MODULES: |
| 64 | + ignored_names.add(alias.name) |
| 65 | + else: |
| 66 | + imports.append(f' import {alias.name}') |
| 67 | + |
| 68 | + def _process_import_from( |
| 69 | + self, node: ast.ImportFrom, imports: list[str], classes: set[str], ignored_names: set[str] |
| 70 | + ) -> None: |
| 71 | + """Process 'from ... import' statements.""" |
| 72 | + module = node.module or '' |
| 73 | + if any(ignored in module for ignored in self.IGNORED_MODULES): |
| 74 | + for alias in node.names: |
| 75 | + ignored_names.add(alias.name) |
| 76 | + else: |
| 77 | + for alias in node.names: |
| 78 | + imports.append(f' from {module} import {alias.name}') |
| 79 | + if alias.name[0].isupper(): # Assume uppercase names are classes |
| 80 | + classes.add(alias.name) |
| 81 | + |
| 82 | + def _process_method_call( |
| 83 | + self, node: ast.Call, method_calls: dict[str, set[str]], ignored_names: set[str] |
| 84 | + ) -> None: |
| 85 | + """Process method calls like obj.method().""" |
| 86 | + if isinstance(node.func.value, ast.Name): |
| 87 | + obj_name = node.func.value.id |
| 88 | + method_name = node.func.attr |
| 89 | + if obj_name not in ignored_names: |
| 90 | + if obj_name not in method_calls: |
| 91 | + method_calls[obj_name] = set() |
| 92 | + method_calls[obj_name].add(method_name) |
| 93 | + |
| 94 | + def create_validation_script( |
| 95 | + self, example_dir: Path, imports: list[str], classes: set[str] |
| 96 | + ) -> str: |
| 97 | + """Generate Python script that validates framework classes can be imported.""" |
| 98 | + import_lines = '\n'.join(f' {imp.strip()}' for imp in imports) |
| 99 | + class_checks = '\n'.join( |
| 100 | + f" assert hasattr({cls}, '__init__'), '{cls} missing __init__'" |
| 101 | + for cls in sorted(classes) |
| 102 | + ) |
| 103 | + |
| 104 | + template = """ |
| 105 | +import sys |
| 106 | +sys.path.insert(0, r"{example_dir}") |
| 107 | +try: |
| 108 | +{imports} |
| 109 | +
|
| 110 | +{checks} |
| 111 | + print("SUCCESS") |
| 112 | +except Exception as e: |
| 113 | + print(f"ERROR: {{e}}") |
| 114 | + sys.exit(1)""" |
| 115 | + |
| 116 | + return template.format(example_dir=example_dir, imports=import_lines, checks=class_checks) |
| 117 | + |
| 118 | + def run_in_isolated_env(self, script: str, example_dir: Path) -> None: |
| 119 | + """Execute validation script using uv in the framework's environment.""" |
| 120 | + try: |
| 121 | + # Run script in framework's isolated environment with its dependencies |
| 122 | + subprocess.run( |
| 123 | + ['uv', 'run', 'python', '-c', script], |
| 124 | + cwd=example_dir, |
| 125 | + capture_output=True, |
| 126 | + text=True, |
| 127 | + check=True, |
| 128 | + ) |
| 129 | + except subprocess.CalledProcessError as e: |
| 130 | + pytest.fail(f'API shape validation failed: {e.stderr.strip()}') |
| 131 | + except FileNotFoundError: |
| 132 | + pytest.skip('uv command not found - please install uv') |
0 commit comments