-
Notifications
You must be signed in to change notification settings - Fork 0
Description
Context
Sub-issue of getaxonflow/axonflow-enterprise#1128.
Users integrating AxonFlow with LangGraph's MultiServerMCPClient need to wrap every MCP tool call with AxonFlow policy enforcement. The pattern is:
mcp_check_input → handler() → mcp_check_output
AxonFlowLangGraphAdapter is already the canonical LangGraph integration surface in the SDK, and users who reach for it are exactly the audience who will also be using MultiServerMCPClient. This pattern should be a first-class feature of the adapter rather than something users have to wire up manually.
Proposed Change
Add an mcp_tool_interceptor() factory method to AxonFlowLangGraphAdapter that returns an async callable ready to pass directly to MultiServerMCPClient:
mcp_client = MultiServerMCPClient(
{"lookup": {"url": "...", "transport": "http"}},
tool_interceptors=[adapter.mcp_tool_interceptor()],
)The returned interceptor should:
- Derive
connector_typefrom the incoming request - Call
self.client.mcp_check_input(...)— raisePolicyViolationErrorif blocked - Call
handler(request)to execute the tool - Call
self.client.mcp_check_output(...)— raisePolicyViolationErrorif hard-blocked; returnredacted_datain place of the original result if redaction was applied - Return the (possibly redacted) result
Design Decisions to Incorporate
1. Redacted output passthrough
A naive implementation raises on block but ignores mcp_check_output.redacted_data. The adapter should substitute redacted output when present:
result = await handler(request)
output_check = await self.client.mcp_check_output(
connector_type=connector_type,
message=f"{{result: {result!r}}}",
)
if not output_check.allowed:
raise PolicyViolationError(output_check.block_reason or "Tool result blocked by policy")
if output_check.redacted_data is not None:
return output_check.redacted_data # return sanitised version
return result2. Pluggable connector type derivation
The default connector type can be derived as f"{request.server_name}.{request.name}", but different tenants may key connector types differently. The factory method should accept an optional callable to override this:
def mcp_tool_interceptor(
self,
connector_type_fn: Callable[[Any], str] | None = None,
) -> Callable:
def default_connector_type(request) -> str:
return f"{request.server_name}.{request.name}"
resolve = connector_type_fn or default_connector_type
...No New Runtime Dependencies
MCPToolCallRequest from langchain-mcp-adapters is only needed for the type hint. The returned callable should duck-type request.server_name, request.name, and request.args, using TYPE_CHECKING-gated imports for the annotation only — consistent with how the adapter currently avoids importing LangGraph types at runtime.