【自作 MCP】Amazon Bedrock Guardrails の ApplyGuardrail API を MCP 経由で実行し EDR っぽく使ってみた

"""
Bedrock Guardrails MCP Server implementation.
"""

import logging
import os
import sys
import boto3
from typing import Optional
from pydantic import BaseModel, Field
from mcp.server.fastmcp import Context, FastMCP

# Logging
logging.basicConfig(
    level=os.getenv("FASTMCP_LOG_LEVEL", "INFO"),
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger(__name__)

# Bedrock Runtime Client
aws_region = os.environ.get("AWS_REGION", "us-east-1")
try:
    if aws_profile := os.environ.get("AWS_PROFILE"):
        bedrock_client = boto3.Session(
            profile_name=aws_profile, region_name=aws_region
        ).client("bedrock-runtime")
    else:
        bedrock_client = boto3.Session(region_name=aws_region).client("bedrock-runtime")
except Exception as e:
    logger.error(f"Error creating bedrock runtime client: {str(e)}")
    raise

# MCP Response Model
class McpGuardrailFilterResponse(BaseModel):
    filtered_text: str = Field(description="Guardrail適用後のテキスト")
    guardrail_intervened: bool = Field(description="ガードレールが介入したかどうか")
    trace: Optional[dict] = Field(default=None, description="ガードレールのトレース情報")

# FastMCPサーバー初期化
mcp = FastMCP(
    "bedrock-guardrails-mcp-server",
    instructions="""
# Bedrock Guardrails MCP Server

このMCPサーバーは、Amazon Bedrock Guardrailsを使ってテキストのフィルタリングを行うツールを提供します。
""",
    dependencies=[
        "pydantic",
        "boto3",
    ],
)

@mcp.tool(
    name="contents_filter",
    description="Bedrock Guardrailsを使って入力テキストにガードレールを適用し、フィルタ済みテキストを返します"
)
async def contents_filter(
    ctx: Context,
    input_text: str = Field(description="フィルタ対象のテキスト"),
    guardrail_id: Optional[str] = Field(default=None, description="ガードレールID（未指定時は環境変数から取得）"),
    guardrail_version: Optional[str] = Field(default=None, description="ガードレールバージョン（未指定時は環境変数から取得）"),
) -> McpGuardrailFilterResponse:
    """
    Bedrock Guardrailsを使って入力テキストにガードレールを適用し、フィルタ済みテキストを返します。
    """
    try:
        guardrail_id = guardrail_id or os.getenv("GUARDRAIL_ID")
        guardrail_version = guardrail_version or os.getenv("GUARDRAIL_VERSION")
        if not guardrail_id or not guardrail_version:
            raise ValueError("GUARDRAIL_IDとGUARDRAIL_VERSIONは必須です（引数または環境変数で指定）")
        # Guardrails適用用のmessages構築
        logger.info(f"Applying guardrail: id={guardrail_id}, version={guardrail_version}")
        logger.info(f"input_text type: {type(input_text)}, value: {input_text}")
        response = bedrock_client.apply_guardrail(
            content=[
                {
                    "text": {
                        "text": str(input_text)
                    }
                }
            ],
            guardrailIdentifier=guardrail_id,
            guardrailVersion=guardrail_version,
            source="INPUT"
        )

        outputs = response.get("outputs", [])
        filtered_text = outputs[0]["text"] if outputs and "text" in outputs[0] else input_text
        guardrail_intervened = response.get("action", "") == "GUARDRAIL_INTERVENED"
        return {
            "filtered_text": filtered_text,
            "guardrail_intervened": guardrail_intervened,
            "trace": None
        }
    except Exception as e:
        logger.error(f"Error in contents_filter: {str(e)}")
        await ctx.error(f"Guardrails適用時にエラーが発生しました: {str(e)}")
        raise

if __name__ == "__main__":
    mcp.run(transport="stdio")

uv init

# Create virtual environment and activate it
uv venv
source .venv/bin/activate

# Install dependencies
uv add "mcp[cli]" boto3 pydantic

{
  "mcpServers": {
    "bedrock-guardrails-mcp-server": {
      "command": "uv",
      "args": [
        "--directory",
        "ファイル置き場の絶対パス/amazon-bedrock-guardrails-mcp",
        "run",
        "main.py"
      ],
      "env": {
        "FASTMCP_LOG_LEVEL": "ERROR",
        "GUARDRAIL_ID": "rw2shyzgmcp6",
        "GUARDRAIL_VERSION": "DRAFT",
        "AWS_REGION": "ap-northeast-1",
        "AWS_ACCESS_KEY_ID": "ASIAAAAAAAAAAAAAAAAA",
        "AWS_SECRET_ACCESS_KEY": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
      },
      "disabled": false,
      "autoApprove": [],
      "alwaysAllow": []
    }
  }
}

     if not isinstance(result, str):
         try:
+             result = json.dumps(pydantic_core.to_jsonable_python(result), ensure_ascii=False)
-             result = json.dumps(pydantic_core.to_jsonable_python(result))