feat: add mcp_check_input and mcp_check_output methods (#103)

* feat: add mcp_check_input and mcp_check_output methods

Add standalone policy-check methods for external orchestrators to use
AxonFlow as a policy gate. Includes async methods, sync wrappers, and
Pydantic request/response types. 403 responses are treated as valid
policy-blocked results, not errors.

Refs: getaxonflow/axonflow-enterprise#1258

* fix: apply ruff formatting to sync wrappers

* docs: add MCP policy-check endpoints to changelog

* docs: set v3.7.0 release date in changelog

* docs: improve v3.7.0 changelog entry

Author: saurabhjain1592

CHANGELOG.md

@@ -5,6 +5,19 @@ All notable changes to the AxonFlow Python SDK will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [3.7.0] - 2026-02-28
+
+### Added
+
+- **MCP Policy-Check Endpoints** (Platform v4.6.0+): Standalone policy validation for external orchestrators (LangGraph, CrewAI) to enforce AxonFlow policies without executing connector queries
+  - `mcp_check_input(connector_type, statement)`: Validate SQL/commands against input policies (SQLi detection, dangerous query blocking, PII in queries, dynamic policies). Returns `allowed=True` or raises with `block_reason`
+  - `mcp_check_output(connector_type, response_data)`: Validate MCP response data against output policies (PII redaction, exfiltration limits, dynamic policies). Returns original or redacted data with `policy_info`
+  - New types: `MCPCheckInputRequest`, `MCPCheckInputResponse`, `MCPCheckOutputRequest`, `MCPCheckOutputResponse`
+  - Async methods with sync wrappers (`mcp_check_input_sync`, `mcp_check_output_sync`)
+  - Supports both query-style (`response_data`) and execute-style (`message` + `metadata`) output validation
+
+---
+
 ## [3.6.0] - 2026-02-22
 
 ### Added

axonflow/__init__.py

@@ -152,6 +152,10 @@
     ListExecutionsResponse,
     ListUsageRecordsOptions,
     ListWebhooksResponse,
+    MCPCheckInputRequest,
+    MCPCheckInputResponse,
+    MCPCheckOutputRequest,
+    MCPCheckOutputResponse,
     MediaAnalysisResponse,
     MediaAnalysisResult,
     MediaContent,
@@ -240,6 +244,11 @@
     "ConnectorMetadata",
     "ConnectorInstallRequest",
     "ConnectorResponse",
+    # MCP Policy Check types
+    "MCPCheckInputRequest",
+    "MCPCheckInputResponse",
+    "MCPCheckOutputRequest",
+    "MCPCheckOutputResponse",
     # Planning types
     "PlanStep",
     "PlanResponse",

axonflow/client.py

@@ -151,6 +151,8 @@
     ListExecutionsResponse,
     ListUsageRecordsOptions,
     ListWebhooksResponse,
+    MCPCheckInputResponse,
+    MCPCheckOutputResponse,
     MediaContent,
     MediaGovernanceConfig,
     MediaGovernanceStatus,
@@ -1076,6 +1078,110 @@ async def mcp_execute(
         """
         return await self.mcp_query(connector, statement, options)
 
+    async def mcp_check_input(
+        self,
+        connector_type: str,
+        statement: str,
+        operation: str = "query",
+        parameters: dict[str, Any] | None = None,
+    ) -> MCPCheckInputResponse:
+        """Validate an MCP request against configured policies without executing it.
+
+        Use this when an external orchestrator (e.g., LangGraph, CrewAI) manages MCP
+        execution but needs AxonFlow policy enforcement as a pre-execution gate.
+
+        Args:
+            connector_type: Type of MCP connector (e.g., "postgres", "snowflake").
+            statement: The SQL query or command to validate.
+            operation: Operation type - "query" (default) or "execute".
+            parameters: Optional query parameters.
+
+        Returns:
+            MCPCheckInputResponse with allowed status, block reason, and policy info.
+
+        Raises:
+            ConnectorError: If the request fails (non-403 errors only).
+        """
+        url = f"{self._config.endpoint}/api/v1/mcp/check-input"
+        body: dict[str, Any] = {
+            "connector_type": connector_type,
+            "statement": statement,
+            "operation": operation,
+        }
+        if parameters:
+            body["parameters"] = parameters
+
+        if self._config.debug:
+            self._logger.debug(
+                "MCP check-input",
+                connector_type=connector_type,
+                statement=statement[:50],
+            )
+
+        response = await self._http_client.post(url, json=body)
+        data = response.json()
+
+        if not response.is_success and response.status_code != 403:  # noqa: PLR2004
+            error_msg = data.get("error", "MCP check-input failed")
+            raise ConnectorError(error_msg, connector_type, "check-input")
+
+        return MCPCheckInputResponse(**data)
+
+    async def mcp_check_output(
+        self,
+        connector_type: str,
+        response_data: list[dict[str, Any]] | None = None,
+        message: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        row_count: int = 0,
+    ) -> MCPCheckOutputResponse:
+        """Validate MCP response data against configured policies.
+
+        Use this when an external orchestrator manages MCP execution but needs AxonFlow
+        policy enforcement as a post-execution gate (PII redaction, exfiltration limits).
+
+        Args:
+            connector_type: Type of MCP connector (e.g., "postgres", "snowflake").
+            response_data: Array of row objects from a query response.
+            message: Execute-style response message (e.g., "5 rows affected").
+            metadata: Connector metadata for SQLi scanning.
+            row_count: Total number of rows returned.
+
+        Returns:
+            MCPCheckOutputResponse with allowed status, redacted data, and policy info.
+
+        Raises:
+            ConnectorError: If the request fails (non-403 errors only).
+        """
+        url = f"{self._config.endpoint}/api/v1/mcp/check-output"
+        body: dict[str, Any] = {
+            "connector_type": connector_type,
+        }
+        if response_data is not None:
+            body["response_data"] = response_data
+        if message is not None:
+            body["message"] = message
+        if metadata is not None:
+            body["metadata"] = metadata
+        if row_count > 0:
+            body["row_count"] = row_count
+
+        if self._config.debug:
+            self._logger.debug(
+                "MCP check-output",
+                connector_type=connector_type,
+                row_count=row_count,
+            )
+
+        response = await self._http_client.post(url, json=body)
+        data = response.json()
+
+        if not response.is_success and response.status_code != 403:  # noqa: PLR2004
+            error_msg = data.get("error", "MCP check-output failed")
+            raise ConnectorError(error_msg, connector_type, "check-output")
+
+        return MCPCheckOutputResponse(**data)
+
     async def generate_plan(
         self,
         query: str,
@@ -5843,6 +5949,33 @@ def mcp_execute(
         """Execute a statement against an MCP connector (alias for mcp_query)."""
         return self._run_sync(self._async_client.mcp_execute(connector, statement, options))
 
+    def mcp_check_input(
+        self,
+        connector_type: str,
+        statement: str,
+        operation: str = "query",
+        parameters: dict[str, Any] | None = None,
+    ) -> MCPCheckInputResponse:
+        """Validate an MCP request against configured policies without executing it."""
+        return self._run_sync(
+            self._async_client.mcp_check_input(connector_type, statement, operation, parameters)
+        )
+
+    def mcp_check_output(
+        self,
+        connector_type: str,
+        response_data: list[dict[str, Any]] | None = None,
+        message: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        row_count: int = 0,
+    ) -> MCPCheckOutputResponse:
+        """Validate MCP response data against configured policies."""
+        return self._run_sync(
+            self._async_client.mcp_check_output(
+                connector_type, response_data, message, metadata, row_count
+            )
+        )
+
     def generate_plan(
         self,
         query: str,

axonflow/types.py

@@ -358,6 +358,45 @@ def was_redacted(self) -> bool:
         return self.redacted
 
 
+class MCPCheckInputRequest(BaseModel):
+    """Request to validate input against MCP policies."""
+
+    connector_type: str
+    statement: str
+    parameters: dict[str, Any] | None = Field(default=None)
+    operation: str = Field(default="query")
+
+
+class MCPCheckInputResponse(BaseModel):
+    """Result of input policy evaluation."""
+
+    allowed: bool
+    block_reason: str | None = Field(default=None)
+    policies_evaluated: int = Field(default=0, ge=0)
+    policy_info: ConnectorPolicyInfo | None = Field(default=None)
+
+
+class MCPCheckOutputRequest(BaseModel):
+    """Request to validate output against MCP policies."""
+
+    connector_type: str
+    response_data: list[dict[str, Any]] | None = Field(default=None)
+    message: str | None = Field(default=None)
+    metadata: dict[str, Any] | None = Field(default=None)
+    row_count: int = Field(default=0, ge=0)
+
+
+class MCPCheckOutputResponse(BaseModel):
+    """Result of output policy evaluation."""
+
+    allowed: bool
+    block_reason: str | None = Field(default=None)
+    redacted_data: Any | None = Field(default=None)
+    policies_evaluated: int = Field(default=0, ge=0)
+    exfiltration_info: ExfiltrationCheckInfo | None = Field(default=None)
+    policy_info: ConnectorPolicyInfo | None = Field(default=None)
+
+
 class PlanStep(BaseModel):
     """A step in a multi-agent plan."""