Building a Production-Ready AI Content Moderation System with Google ADK and Model Context Protocol (MCP)

How to get free OpenAI moderation service for Google AI agent using MCP.

Image Generated by Canva

Introduction: When Good AI Chats Go Bad

Picture this scenario: You’ve just launched your revolutionary AI chatbot using Google’s Agent Development Kit (ADK). Users are flocking to it, conversations are flowing, and everything seems perfect — until it’s not.

A curious user asks your AI how to build an explosive device. Another probes for inappropriate content. Someone else tests the boundaries with harmful language. Suddenly, your innovative AI assistant has morphed from helpful tool to potential liability.

As AI developers working with Google’s ADK, we face unique challenges. The ADK enables us to create powerful, context-aware agents, but without proper guardrails, these agents can inadvertently produce harmful, dangerous, or simply inappropriate responses.

In this article, I’ll walk you through a practical solution: a production-ready content moderation system built with Google’s Agent Development Kit (ADK) and Model Context Protocol (MCP) that leverages OpenAI’s moderation API. The best part? The moderation API is free to use, making robust content safety accessible to projects of all sizes, while the ADK provides a structured, enterprise-ready framework for your agent development.

The Problem: Unmoderated AI is a Liability

Without proper content moderation, AI systems can:

• Generate harmful, dangerous, or illegal content
• Expose organizations to significant legal and reputational risks
• Create unsafe experiences for users
• Turn promising applications into PR nightmares

For any AI system deployed in production, content moderation isn’t just a nice-to-have — it’s essential.

The Solution: An Integrated Content Moderation Pipeline

My solution provides a clean architecture that:

1. Intercepts every user input to this Google ADK agent
2. Checks it through OpenAI’s moderation API
3. Only passes safe content to the AI model
4. Provides detailed feedback when content is blocked

Let’s build this together!

System Architecture

I’ve implemented two variants of this moderation flow, each with different transport mechanisms that work seamlessly with Google ADK:

SSE Transport Architecture

Server-Sent Events (SSE) is a standard HTTP-based technology where the server can push data to the client. In my implementation, the moderation server runs as a separate service that the ADK agent communicates with over HTTP. This approach is ideal for cloud-based deployments on Google Cloud Run or Google Kubernetes Engine.

Image Generated by Author

STDIO Transport Architecture

Standard I/O (STDIO) transport runs the moderation server in the same process as the client, communicating through standard input/output streams. This creates a more streamlined setup with fewer moving parts, perfect for containerized deployments on Google Cloud’s infrastructure.

Image Generated by Author

Flow Diagrams

Let’s also visualize how the system handles both safe and unsafe content:

Safe Content Flow

Image Generated by Author

Unsafe Content Flow

Image Generated by Author

Key Components

Let’s break down each component of my system and examine the most important code snippets. The code shown here omits some error handling for simplicity. Full code for this project is available on GitHub.

1. Moderation Server

The moderation server exposes a function to moderate content using OpenAI’s API through the Google ADK’s MCP framework:

class ModerationServer:
    """Class for the MCP moderation server."""

    def __init__(self, service_name="Content Moderation Service"):
        """Initialize the MCP server."""
        # OpenAI API key should be set as an environment variable
        self.api_key = os.environ.get("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("OPENAI_API_KEY environment variable is not set")

        # Initialize OpenAI client
        self.client = openai.OpenAI(api_key=self.api_key)

        # Initialize MCP server
        self.mcp = FastMCP(service_name)

        # Register tools
        self.register_tools()

    def register_tools(self):
        """Register the moderation tool with the MCP server."""
        @self.mcp.tool()
        def moderate_content(content: str) -> ModerationResult:
            """Moderate content using OpenAI's moderation API."""
            return self.perform_moderation(content)

    def perform_moderation(self, content: str) -> ModerationResult:
        """Perform content moderation using OpenAI's API."""
        try:
            # Call OpenAI's moderation API
            response = self.client.moderations.create(input=content, model="omni-moderation-latest")

            # Process response
            result = response.results[0]

            return ModerationResult(
                flagged=result.flagged,
                categories={k: v for k, v in result.categories.model_dump().items()},
                category_scores={k: v for k, v in result.category_scores.model_dump().items()}
            )
        except Exception as e:
            # In case of error, return a flagged result with error info
            return ModerationResult(
                flagged=True,
                categories={"error": True},
                category_scores={"error": 1.0}
            )

This code might look simple, but there’s a lot happening:

1. We initialize the OpenAI client with an API key
2. We register a moderate_content tool with the MCP server
3. The moderation function calls OpenAI’s moderation API
4. We process the response and return a structured result
5. We include error handling to fail safely (default to flagging content when errors occur)

This integration with Google’s FastMCP provides a standardized way to expose tools to ADK agents, making the moderation service discoverable and usable by any ADK-based application.

2. Moderation Tool

The moderation tool is the client-side wrapper that communicates with the moderation server:

class ModerationTool:
    """Wrapper class for the MCP moderation tool."""

    def __init__(self, mcp_tools):
        """Initialize the moderation tool with MCP tools."""
        self.mcp_tools = mcp_tools
        # Find the moderate_content tool
        self.moderate_content_tool = next(
            (tool for tool in mcp_tools if tool.name == "moderate_content"), 
            None
        )
        if not self.moderate_content_tool:
            raise ValueError("'moderate_content' tool not found in MCP server!")

    async def moderate(self, content: str) -> Dict[str, Any]:
        """Moderate content using the MCP tool."""
        try:
            # Prepare arguments for run_async
            args = {"content": content}
            tool_context = {
                "user_id": "user123",
                "session_id": "session456",
                "request_id": "req789"
            }

            # Make a single request to the MCP server
            raw_result = await self.moderate_content_tool.run_async(
                args=args,
                tool_context=tool_context
            )

            # Process the CallToolResult object
            result = self.extract_result_data(raw_result)

            if result:
                return result
            else:
                return {
                    "error_message": "Failed to extract moderation result from response"
                }

        except Exception as e:
            print(f"Error in moderation: {e}")
            traceback.print_exc()
            return {
                "error_message": f"Error in moderation: {e}"
            }

This tool leverages Google ADK’s asynchronous tooling capabilities, making it efficient and non-blocking — perfect for production deployments.

3. Moderation Agent

The moderation agent is where everything comes together. It intercepts user queries, sends them to the moderation tool, and only forwards safe content to the LLM:

async def process_query(self, user_input: str) -> str:
    """Process a user query with moderation check."""
    try:
        # Check content moderation first
        print(f"Checking moderation for: '{user_input}'")
        moderation_result = await self.moderation_tool.moderate(user_input)
        print("Moderation check completed.")

        # Extract flagged status and categories
        flagged = False
        categories = {}

        # Direct access to flagged and categories
        if isinstance(moderation_result, dict):
            if "flagged" in moderation_result:
                flagged = moderation_result["flagged"]
                categories = moderation_result.get("categories", {})
            # Check for result key
            elif "result" in moderation_result:
                result = moderation_result["result"]
                if isinstance(result, dict) and "flagged" in result:
                    flagged = result["flagged"]
                    categories = result.get("categories", {})

        # Handle flagged content
        if flagged:
            # Content was flagged, prepare explanation
            flagged_categories = {k: v for k, v in categories.items() if v}
            response = f"Your query contains content that violates our content policies.\nFlagged categories: {', '.join(flagged_categories.keys() or ['unknown'])}"
            return response

        # Content is safe, process with agent runner
        print("Content is safe, processing with agent runner...")
        return await self._use_agent_runner(user_input)

    except Exception as e:
        traceback_str = traceback.format_exc()
        print(f"Exception traceback: {traceback_str}")
        return f"Error during processing: {e}"

This is the critical safety check: if the content is flagged by the moderation API, we immediately return an explanation to the user without ever sending the query to the ADK agent’s underlying LLM.

Real-World Implementation

Now let’s see how it all works in practice. The system supports two transport methods (SSE and STDIO), each with its own advantages:

Image Generated by Author

When to Use Each Transport Method

• Use SSE when you need to deploy the moderation server separately from your agent, perhaps to serve multiple agents or for better scaling.
• Use STDIO for simpler deployments where everything runs on the same machine, reducing network overhead and complexity.

Testing the System

The testing framework in this project is designed to thoroughly validate all components of the moderation system. The test_moderation_system.py script runs a comprehensive set of tests that verifies safe queries, unsafe queries and queries provided by a user.

Taking it to Production

While this system is already quite robust, here are some enhancements you’d want to make for a full production deployment:

1. Authentication: Add proper authentication for the MCP server
2. Containerization: Package applications as Docker containers
3. Monitoring: Implement health checks and metrics collection
4. Logging: Enhance logging for better troubleshooting
5. Rate Limiting: Prevent API abuse
6. Caching: Add caching for repetitive moderation requests
7. Failover Mechanisms: Implement retry logic and fallbacks
8. Security Hardening: Review and address security concerns.

Deployment Options

The system is flexible enough to be deployed in various environments:

Cloud Deployment Options

1. Containerized Services

Package the components as Docker containers and orchestrate with Kubernetes for scalability and portability.

2. Serverless Deployments

For event-driven scenarios, consider serverless functions like Google Cloud Functions.

3. Cloud-Managed Hosting

Services like Google Cloud Run can host your MCP servers with minimal infrastructure management.

Self-Hosted Deployment Options

1. On-Premises Servers

Install directly on physical servers in your data center for maximum control.

2. Private Cloud

Deploy on private cloud infrastructure for enhanced security while maintaining virtualization benefits.

3. Edge Deployment

For latency-sensitive applications, deploy at the edge closer to where data is generated.

Google ADK Integration Benefits

The Google Agent Development Kit provides several key benefits for this moderation system:

1. Structured Agent Framework: ADK’s LlmAgent class provides a robust foundation for building agents with clear instruction handling and response formatting.
2. Built-in Tool Support: The ability to easily add tools to agents makes integrating moderation seamless.
3. Session Management: ADK’s session services allow for maintaining context across interactions, crucial for tracking user behavior.
4. MCP Integration: The Model Context Protocol enables standardized communication between the agent and moderation service.
5. Asynchronous Processing: ADK’s support for async operations ensures the moderation system doesn’t block the main application flow.
6. Enterprise Readiness: ADK’s architecture follows Google’s best practices for production-grade AI systems.

Real-World Applications with Google ADK

This moderation system isn’t just a technical demo — it solves real business problems:

• Educational Platforms: Ensure student interactions with AI tutors remain appropriate and safe
• Customer Service Bots: Prevent abuse while maintaining helpful service
• Content Generation Tools: Filter requests for inappropriate creative content
• Internal Enterprise Tools: Maintain professional standards in workplace AI usage
• Healthcare Chatbots: Prevent requests for harmful medical advice
• Social Applications: Screen user-generated prompts in AI-powered social features

Conclusion: Safety First, Not Safety Last

Content moderation is often treated as an afterthought in AI projects, but it should be considered from day one. The system we’ve built demonstrates that with modern tools like Google ADK and MCP, implementing robust content moderation doesn’t have to be complicated or expensive.

By deploying on Google Cloud and leveraging OpenAI’s free moderation API, even small projects can ensure their AI applications act responsibly and safely. The dual-transport architecture we’ve designed offers flexibility for different deployment scenarios, from simple single-machine setups to complex distributed systems on Google’s global infrastructure.

Remember, the best AI applications aren’t just the ones with the most impressive capabilities — they’re the ones that earn users’ trust by consistently behaving responsibly. Content moderation is a key part of building that trust, and Google ADK provides the perfect framework to implement it.

Next Steps

Ready to build on this foundation? Here are some exciting directions to explore:

1. Explanation Generation: Provide human-readable explanations for moderation decisions to improve accuracy for Human-in-the-Loop monitoring, allowing reviewers to quickly understand why content was flagged and make better override decisions
2. Web UI: Add a dashboard for testing and monitoring moderation
3. Feedback Loop: Learn from false positives/negatives
4. Logging: Implement Google Cloud Logging for enhanced audit trails
5. MCP Security Hardening: Review and address security concerns in the MCP communication layer, ensuring robust authentication, encryption, and proper permission models for production environments
6. Testing: Implement A/B testing with Google Cloud’s experimentation framework

The code for this project is available on GitHub, and I’d love to hear how you adapt it for your own Google ADK applications! Feel free to connect on LinkedIn to continue the discussion!

Have you implemented content moderation in your AI systems?

What approaches have worked best for you? Let me know in the comments!

Happy (safe!) AI developing with Google ADK! 🚀✨