Validate MCP tool composition correctness and safety

Introduction

When you chain MCP tools into multi-step workflows, a single type mismatch between one tool's output and the next tool's input can silently corrupt data—an array of objects flowing into a slot that expects an array of strings, or a nullable field landing in a required integer slot. Teams that defer composition validation to runtime discover these structural defects only after production incidents, when the blast radius is largest. By the end of this lesson you'll be able to validate MCP tool compositions for structural type compatibility across field mappings, before any token is issued for execution.

Key Terminology

• Composition DAG: The directed acyclic graph representation of a multi-tool workflow where nodes are MCP tool invocations and edges are typed data-flow connections between them.
• CompositionEdge: A dataclass that models one connection in the composition DAG, naming the source tool, the target tool, and a field_mappings dictionary that declares which source output field flows into which target input field.
• Field Mapping: A declared correspondence between an output field on one tool and an input field on the next, against which structural type compatibility is checked.
• SchemaCompatibilityChecker: The component that performs structural subtyping on two JSON Schemas, recursively comparing types, required properties, and item schemas to decide whether a source can satisfy a target.
• ToolCompositionValidator: The driver class that walks every CompositionEdge, fetches input and output schemas from the tool registry, and delegates each field-mapping check to the SchemaCompatibilityChecker.
• ValidationIssue: A dataclass that captures one structural problem discovered during validation, including severity (ERROR or WARNING), the source and target tool names, a human-readable message, and the field_path where the mismatch occurs.
• Numeric Widening: A JSON Schema compatibility rule that permits an integer source to satisfy a number target without producing a type-mismatch error.
• Composition Token: An OAuth 2.1 access token whose scope covers every tool in a validated pipeline, issued only after structural type validation passes on every edge.

Concepts

Why Individual Tool Validation Is Insufficient

Each MCP server publishes its tools with JSON Schema definitions for inputs and outputs. The tool registry stores these schemas, and capability discovery exposes them to clients. However, these schemas describe tools in isolation. Consider a three-tool pipeline:

1. db_query — outputs {"type": "object", "properties": {"rows": {"type": "array", "items": {"type": "object"}}}}
2. transform_csv — expects {"type": "object", "properties": {"data": {"type": "array", "items": {"type": "string"}}}}
3. email_send — expects {"type": "object", "properties": {"body": {"type": "string"}, "to": {"type": "string"}}}

The db_query tool outputs an array of objects, but transform_csv expects an array of strings. This mismatch compiles to None at the individual tool level—both schemas are independently valid—but the composition fails at runtime when the transformer receives nested objects instead of flat strings. Worse, some languages and serializers silently coerce types, producing garbled output rather than a clean error. Composition validation catches these mismatches at definition time, before any tool executes.

Handling Edge Cases in Production

Several composition patterns require special handling that the basic validator must account for:

• Optional intermediate tools: When a composition includes conditional branches (tool B executes only if tool A returns a specific status), the validator must check type compatibility on all possible paths, not just the happy path. Fields that might be None in the source schema must be matched against nullable targets, or the validator must emit a WARNING that the target tool may receive None for a required field.

• Fan-out compositions: A single tool's output feeds multiple downstream tools simultaneously. The validator must check type compatibility on each outbound edge independently, because the same source field may need to satisfy different target schemas with different required-property sets or item types.

• Schema evolution: When the tool registry updates a tool's output schema (a new field is added, or a field's type changes from integer to number), all compositions referencing that tool must be revalidated. The ecosystem governance dashboard tracks composition freshness—the time since last validation relative to the last schema change for any constituent tool—and flags stale compositions for re-validation.

• Recursive compositions: A composition that references another composition as a sub-pipeline must be flattened before validation. The validator detects cycles by maintaining a visited-set during traversal and rejects any composition that references itself directly or transitively, returning an ERROR with the cycle path for debugging.

Testing Composition Validation

Effective testing of the composition validator requires two categories of structural test cases. First, positive type tests confirm that compatible schemas (e.g., integer source with number target, or a source object that provides all of a target's required properties) produce zero errors. Second, negative type tests verify that incompatible schemas (e.g., string source with integer target, an array of objects feeding a slot expecting an array of strings, or a missing required property) produce specific, actionable error messages including the full field path. Each test should construct a minimal composition DAG, invoke validation, and assert on the exact ValidationIssue list, checking severity, tool names, and message content. This approach ensures that as new tools are added to the registry and as existing tool schemas evolve, the validator continues to catch structural drift across the entire tool mesh.

Code Walkthrough

The Composition Graph Model

A tool composition is a directed acyclic graph (DAG) where nodes represent MCP tool invocations and edges represent data flow between them. Each edge carries a mapping specification that declares which fields from the source tool's output feed into the destination tool's input. The validator traverses this graph topologically, checking type compatibility at every edge and accumulating structural mismatches into a single report.

Loading diagram...

In this graph, the validator must verify that db_query.rows is type-compatible with transform_csv.data, that db_query.rows is compatible with aggregate_stats.records, and that both paths into email_send produce valid body fields. The fan-in into email_send requires checking each incoming edge independently, because two different source fields (csv_string and summary) must each satisfy the same target schema for body.

Implementing the Type Compatibility Checker

The following implementation defines the core ToolCompositionValidator class alongside a SchemaCompatibilityChecker that performs structural subtyping on JSON Schema types. The SchemaCompatibilityChecker.is_compatible method implements a recursive algorithm that checks whether a source schema can satisfy a target schema's constraints, handling nested objects, arrays with typed items, union types, and nullable fields. The ToolCompositionValidator class consumes composition definitions—lists of CompositionEdge objects—and delegates type checking to the compatibility checker while accumulating validation errors into a structured report. This design separates the concern of schema analysis from pipeline traversal, making each component independently testable against your tool registry's schema catalog.

1from dataclasses import dataclass, field
2from enum import Enum
3
4class Severity(Enum):
5    ERROR = "error"
6    WARNING = "warning"
7
8@dataclass
9class ValidationIssue:
10    severity: Severity
11    source_tool: str
12    target_tool: str
13    message: str
14    field_path: str = ""
15
16@dataclass
17class CompositionEdge:
18    source_tool: str
19    target_tool: str
20    field_mappings: dict[str, str]  # source_field -> target_field
21
22class SchemaCompatibilityChecker:
23    NUMERIC_TYPES = {"integer", "number"}
24    
25    def is_compatible(self, source: dict, target: dict) -> list[str]:
26        errors = []
27        src_type = source.get("type")
28        tgt_type = target.get("type")
29        
30        if tgt_type == "number" and src_type in self.NUMERIC_TYPES:
31            return errors
32        if src_type != tgt_type:
33            errors.append(
34                f"Type mismatch: source={src_type}, target={tgt_type}"
35            )
36            return errors
37        
38        if tgt_type == "object":
39            tgt_props = target.get("properties", {})
40            src_props = source.get("properties", {})
41            tgt_required = set(target.get("required", []))
42            for prop_name in tgt_required:
43                if prop_name not in src_props:
44                    errors.append(f"Missing required property: {prop_name}")
45                else:
46                    nested = self.is_compatible(
47                        src_props[prop_name], tgt_props[prop_name]
48                    )
49                    errors.extend(
50                        f"{prop_name}.{e}" for e in nested
51                    )
52        
53        if tgt_type == "array":
54            src_items = source.get("items", {})
55            tgt_items = target.get("items", {})
56            if src_items and tgt_items:
57                nested = self.is_compatible(src_items, tgt_items)
58                errors.extend(f"items.{e}" for e in nested)
59        
60        return errors

Driving Validation Across the Composition DAG

The ToolCompositionValidator consumes a list of CompositionEdge objects, fetches each tool's input and output schemas from the registry, and iterates over every field mapping to invoke the SchemaCompatibilityChecker. The result is a single list of ValidationIssue objects covering every structural mismatch across the entire pipeline, so authors fix all type defects in one pass rather than discovering them edge-by-edge at runtime.

1class ToolCompositionValidator:
2    def __init__(self, registry_client, checker: SchemaCompatibilityChecker):
3        self._registry = registry_client
4        self._checker = checker
5
6    def validate(self, edges: list[CompositionEdge]) -> list[ValidationIssue]:
7        issues: list[ValidationIssue] = []
8        for edge in edges:
9            src_schema = self._registry.get_output_schema(edge.source_tool)
10            tgt_schema = self._registry.get_input_schema(edge.target_tool)
11            src_props = src_schema.get("properties", {})
12            tgt_props = tgt_schema.get("properties", {})
13            for src_field, tgt_field in edge.field_mappings.items():
14                if src_field not in src_props:
15                    issues.append(ValidationIssue(
16                        severity=Severity.ERROR,
17                        source_tool=edge.source_tool,
18                        target_tool=edge.target_tool,
19                        message=f"Source field missing: {src_field}",
20                        field_path=src_field,
21                    ))
22                    continue
23                if tgt_field not in tgt_props:
24                    issues.append(ValidationIssue(
25                        severity=Severity.ERROR,
26                        source_tool=edge.source_tool,
27                        target_tool=edge.target_tool,
28                        message=f"Target field missing: {tgt_field}",
29                        field_path=tgt_field,
30                    ))
31                    continue
32                errors = self._checker.is_compatible(
33                    src_props[src_field], tgt_props[tgt_field]
34                )
35                for err in errors:
36                    issues.append(ValidationIssue(
37                        severity=Severity.ERROR,
38                        source_tool=edge.source_tool,
39                        target_tool=edge.target_tool,
40                        message=err,
41                        field_path=f"{src_field}->{tgt_field}",
42                    ))
43        return issues

Integrating with OAuth 2.1 Authorization

MCP's OAuth 2.1 authorization framework issues scoped tokens for tool execution. Type-compatibility validation integrates at the token-issuance boundary: before the authorization server grants a composition-scoped token, it invokes the validator. If validation surfaces any ERROR-severity structural mismatch, the token request is denied with a structured error response. This creates an enforcement point that cannot be bypassed—no valid token means no execution, regardless of whether individual tools would authorize independently.

The integration flow works as follows:

• Key Terminology:

• Composition Token: An OAuth 2.1 access token whose scope covers multiple tools in a single pipeline, issued only after structural type validation passes.

• Structural Subtyping: A type compatibility approach where a source schema satisfies a target if it provides at least all required properties with compatible types, regardless of additional properties.

• Composition DAG: The directed acyclic graph representation of a multi-tool workflow where nodes are tool invocations and edges are typed data-flow connections.

• Field Mapping: A declared correspondence between an output field on one tool and an input field on the next, against which structural type compatibility is checked.

• Numeric Widening: A JSON Schema compatibility rule that permits an integer source to satisfy a number target without producing a type-mismatch error.

Before a client can chain MCP tools together, the OAuth 2.1 Auth Server enforces a type-compatibility gate. The ToolCompositionValidator walks every declared field mapping in the composition DAG and asks the SchemaCompatibilityChecker whether the source schema structurally satisfies the target schema. Only when every edge passes does the TokenStore issue a scoped composition token; otherwise the client receives a 403 Forbidden with the list of mismatches, blocking type-incompatible pipelines before they execute.

Loading diagram...

This sequence ensures that type-compatibility validation is not merely advisory—it is a hard gate in the authorization flow. The tool routing layer downstream accepts only composition tokens that embed the validated tool list, preventing runtime substitution of tools whose schemas were never checked against the pipeline's field mappings.

Do's and Don'ts

Do's

1. ✓Do validate field mappings at every CompositionEdge before issuing execution tokens — a structural mismatch between db_query.rows (an array of objects) and a target field expecting an array of strings silently corrupts data at runtime; catching it in SchemaCompatibilityChecker.is_compatible at composition-definition time shrinks the blast radius to a build-time report rather than a production incident.
2. ✓Do treat fan-in edges into a single target field as independent checks — when both transform_csv.csv_string and aggregate_stats.summary map to email_send.body, each incoming edge must be validated separately against the target schema, because a valid path from one source does not guarantee the other source satisfies the same body constraints.
3. ✓Do separate schema analysis (SchemaCompatibilityChecker) from graph traversal (ToolCompositionValidator) — keeping recursive type-compatibility logic (nullable handling, nested object required-property checks, array items subtyping) isolated from DAG edge iteration makes each component independently testable against your tool registry's schema catalog without re-exercising the full pipeline.

Don'ts

1. ✗Don't conflate integer and number type compatibility with general type leniency — SchemaCompatibilityChecker intentionally permits integer → number widening (via NUMERIC_TYPES) but nothing else; adding ad-hoc string-to-number coercions or silently dropping type mismatches in is_compatible would allow structurally broken compositions to pass validation and corrupt downstream tool inputs.
2. ✗Don't skip required-property checks on nested objects in composition edges — the validator iterates target.get("required", []) and flags any property missing from the source schema; omitting this check lets a composition where db_query.rows lacks a field that aggregate_stats.records requires appear valid, only to raise a key error when the actual tool executes.
3. ✗Don't defer composition validation to the first live execution of the DAG — the entire point of topological traversal with accumulated ValidationIssue objects is to surface all structural mismatches (type conflicts, missing required fields, array item schema gaps) in a single pre-execution report; skipping the validator and relying on runtime tool errors means every fan-out branch of the graph must fail independently before the full set of incompatibilities is known.