MCP Security Guide for AI Clients and Agents

import json
from os import environ

from anthale import Anthale

client = Anthale(api_key=environ["ANTHALE_API_KEY"])
MCP_POLICY_ID = environ["ANTHALE_MCP_POLICY_ID"]

ALLOWED_MCP_SERVERS = {
"anthale-docs": {"search_anthale"},
"ticketing": {"create_ticket"},
}

def guard_mcp_call(server_name: str, tool_name: str, approved: bool) -> None:
if tool_name not in ALLOWED_MCP_SERVERS.get(server_name, set()):
raise RuntimeError("This MCP tool is not allowed for the workflow.")

    if server_name == "ticketing" and tool_name == "create_ticket" and not approved:
        raise RuntimeError("Mutating MCP tools require approval.")

def guard_mcp_tool_result(server_name: str, tool_name: str, tool_result: dict) -> dict:
response = client.organizations.policies.enforce(
policy_identifier=MCP_POLICY_ID,
direction="input",
messages=[{"role": "user", "content": json.dumps(tool_result)}],
metadata={
"boundary": "mcp_output",
"feature": "support-agent",
"serverName": server_name,
"toolName": tool_name,
"sourceType": "mcp",
},
)
if response.action == "block":
raise RuntimeError("Anthale blocked the MCP tool result.")

    safe_content = response.redacted_messages[0]["content"] if response.action == "redact" else json.dumps(tool_result)
    return json.loads(safe_content)

import Anthale from "anthale"

const client = new Anthale({ apiKey: process.env.ANTHALE_API_KEY })
const MCP_POLICY_ID = process.env.ANTHALE_MCP_POLICY_ID!

const ALLOWED_MCP_SERVERS = new Map<string, Set<string>>([
  ["anthale-docs", new Set(["search_anthale"])],
  ["ticketing", new Set(["create_ticket"])],
])

export function guardMcpCall(serverName: string, toolName: string, approved: boolean): void {
  if (!ALLOWED_MCP_SERVERS.get(serverName)?.has(toolName)) {
    throw new Error("This MCP tool is not allowed for the workflow.")
  }

  if (serverName === "ticketing" && toolName === "create_ticket" && !approved) {
    throw new Error("Mutating MCP tools require approval.")
  }
}

export async function guardMcpToolResult(
  serverName: string,
  toolName: string,
  toolResult: Record<string, unknown>,
): Promise<Record<string, unknown>> {
  const response = await client.organizations.policies.enforce(MCP_POLICY_ID, {
    direction: "input",
    messages: [{ role: "user", content: JSON.stringify(toolResult) }],
    metadata: {
      boundary: "mcp_output",
      feature: "support-agent",
      serverName,
      toolName,
      sourceType: "mcp",
    },
  })

  if (response.action === "block") {
    throw new Error("Anthale blocked the MCP tool result.")
  }

  const safeContent = response.action === "redact" ? response.redacted_messages[0].content : JSON.stringify(toolResult)
  return JSON.parse(safeContent) as Record<string, unknown>
}