Stop Chatting. Start Acting.

Create a Custom MCP Client

Build a simple Python client with `httpx` to interact with the ScrapingBee MCP server programmatically.

# custom_mcp_client.py
import os
import re
import json
import asyncio
import httpx
import google.generativeai as genai
from typing import Dict, Any, List

# ---------------------------------------------------------------------
# 🔧 Configuration
# ---------------------------------------------------------------------
MCP_SERVER_URL = "https://mcp.scrapingbee.com/mcp"
SCRAPINGBEE_API_KEY = os.environ.get("SCRAPINGBEE_API_KEY")
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")


# ---------------------------------------------------------------------
# ⚙️ MCP Client
# ---------------------------------------------------------------------
class MCPClient:
    """A minimal async client to communicate with an MCP server."""

    def __init__(self, base_url: str, api_key: str):
        self.base_url = f"{base_url}?api_key={api_key}"
        self.session_id = None
        self.http_client = httpx.AsyncClient(timeout=300)
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            "User-Agent": "Minimal-MCP-Client/1.0",
        }

    async def _send_request(self, payload: Dict[str, Any]) -> Dict[str, Any] | None:
        """Send a JSON-RPC request and handle JSON or SSE responses."""
        headers = self.headers.copy()
        if self.session_id:
            headers["Mcp-Session-Id"] = self.session_id

        try:
            response = await self.http_client.post(self.base_url, json=payload, headers=headers)
            response.raise_for_status()

            text = await response.aread()
            text_str = text.decode("utf-8")

            # Detect Server-Sent Events (SSE) and extract JSON
            if "data:" in text_str:
                match = re.search(r"data:\s*({.*})", text_str, re.DOTALL)
                if match:
                    return json.loads(match.group(1))

            # Some notifications may have no body
            if response.status_code in [202, 204] or not text_str:
                return {"status": "ok"}

            return json.loads(text_str)

        except Exception as e:
            print(f"❌ Request error: {e}")
            return None

    async def initialize(self) -> List[Dict[str, Any]] | None:
        """Perform MCP handshake and list available tools."""
        # 1️⃣ Initialize session
        init_payload = {
            "jsonrpc": "2.0",
            "id": "1",
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "MinimalClient"},
            },
        }

        async with self.http_client.stream("POST", self.base_url, json=init_payload, headers=self.headers) as response:
            self.session_id = response.headers.get("Mcp-Session-Id")

        if not self.session_id:
            print("❌ Could not establish MCP session.")
            return None

        print(f"✅ MCP Session established: {self.session_id}")

        # 2️⃣ Send `notifications/initialized`
        notify_payload = {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}
        await self._send_request(notify_payload)

        # 3️⃣ List available tools
        list_payload = {"jsonrpc": "2.0", "id": "2", "method": "tools/list", "params": {}}
        response_data = await self._send_request(list_payload)

        if response_data and "result" in response_data and "tools" in response_data["result"]:
            tool_count = len(response_data["result"]["tools"])
            print(f"🔧 {tool_count} tools fetched successfully from MCP.")
            return response_data["result"]["tools"]

        print("❌ Failed to retrieve tool list.")
        return None

    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any] | None:
        """Execute a specific MCP tool."""
        payload = {
            "jsonrpc": "2.0",
            "id": "3",
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": arguments},
        }
        return await self._send_request(payload)


# ---------------------------------------------------------------------
# 🧩 Gemini Tool Schema Conversion
# ---------------------------------------------------------------------
def format_tools_for_gemini(tools: List[Dict[str, Any]]) -> List[genai.protos.Tool]:
    """
    Converts MCP tool schemas into Gemini FunctionDeclarations,
    allowing Gemini to reason about which tool to call.
    """
    gemini_tools = []
    type_map = {
        "string": genai.protos.Type.STRING,
        "integer": genai.protos.Type.INTEGER,
        "number": genai.protos.Type.NUMBER,
        "boolean": genai.protos.Type.BOOLEAN,
    }

    for tool in tools:
        properties = {}
        input_schema = tool.get("inputSchema", {})

        # Build parameter schema
        if "properties" in input_schema:
            for name, prop in input_schema["properties"].items():
                properties[name] = genai.protos.Schema(
                    type=type_map.get(prop.get("type", "string"), genai.protos.Type.STRING),
                    description=prop.get("description", ""),
                )

        func = genai.protos.FunctionDeclaration(
            name=tool["name"],
            description=tool["description"],
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties=properties,
                required=input_schema.get("required", []),
            ),
        )

        gemini_tools.append(genai.protos.Tool(function_declarations=[func]))

    return gemini_tools


# ---------------------------------------------------------------------
# 🤖 Main Program Flow
# ---------------------------------------------------------------------
async def main():
    """Runs the MCP–Gemini integration demo."""
    if not SCRAPINGBEE_API_KEY or not GEMINI_API_KEY:
        print("❌ Please set SCRAPINGBEE_API_KEY and GEMINI_API_KEY environment variables.")
        return

    # 1️⃣ Connect to MCP and list tools
    mcp_client = MCPClient(MCP_SERVER_URL, SCRAPINGBEE_API_KEY)
    tools = await mcp_client.initialize()
    if not tools:
        return

    # 2️⃣ Configure Gemini with MCP tool schemas
    genai.configure(api_key=GEMINI_API_KEY)
    gemini_tools = format_tools_for_gemini(tools)
    model = genai.GenerativeModel(model_name="gemini-2.5-pro", tools=gemini_tools)
    print("\n✨ Gemini is now configured with MCP tools.")

    # 3️⃣ Get user request and let Gemini pick a tool
    user_prompt = input("\nAsk the AI to perform a task (e.g., 'Search Walmart for iPhones'):\n> ")
    if not user_prompt.strip():
        return

    print("\n🤖 Asking Gemini which MCP tool to use...")
    response = model.generate_content(user_prompt)
    function_call = response.candidates[0].content.parts[0].function_call

    if not function_call.name:
        print("\n💬 Gemini Response (no tool selected):")
        print(response.text)
        return

    tool_name = function_call.name
    tool_args = dict(function_call.args)
    print(f"🧠 Gemini chose tool: '{tool_name}' with arguments: {tool_args}")

    # 4️⃣ Execute Gemini’s selected tool
    print(f"\n⚙️ Running '{tool_name}' on MCP...")
    result = await mcp_client.call_tool(tool_name, tool_args)

    # 5️⃣ Display the result
    print("\n📄 --- MCP Tool Result ---")
    print(json.dumps(result, indent=2) if result else "No result returned.")

# ---------------------------------------------------------------------
# 🚀 Entry Point
# ---------------------------------------------------------------------
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nExiting gracefully.")