Google Custom Search Engine MCP Server

Verified
The Unlicense
Overview InspectNew Schema Related Servers Reviews Score
mcp-google-cse
src
mcp_google_cse
import os
from sys import stdin, stdout
from typing import Any, List

from googleapiclient.discovery import build
from mcp import Tool, stdio_server
from mcp.server.lowlevel import Server
from mcp.types import TextContent, ImageContent, EmbeddedResource

stdin.reconfigure(encoding='utf-8')
stdout.reconfigure(encoding='utf-8')

server = Server("mcp-google-cse")


@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    """
    List available tools.
    Each tool specifies its arguments using JSON Schema validation.
    """
    return [
        Tool(
            name="google_search",
            description="Search the custom search engine using the search term. Regular query arguments can also be used, like appending site:reddit.com or after:2024-04-30. If available and/or requested, the links of the search results should be used in a follow-up request using a different tool to get the full content. Example: \"claude.ai features site:reddit.com after:2024-04-30\"",
            inputSchema={
                "type": "object",
                "properties": {
                    "search_term": {"type": "string"},
                },
                "required": ["search_term"],
            }
        )
    ]


@server.call_tool()
async def handle_call_tool(
        name: str, arguments: dict[str, Any] | None
) -> list[TextContent | ImageContent | EmbeddedResource]:
    """
    Handle tool execution requests.
    Tools can modify server state and notify clients of changes.
    """
    search_term = arguments.get('search_term', '')

    if name == 'google_search' and search_term != '':
        result = await google_search(search_term)
        return result
    else:
        raise ValueError(f"Unknown tool: {name}")


async def google_search(search_term: str) -> List[TextContent]:
    """
    Search the custom search engine using the search term.
    :param search_term: The search term to search for, equaling the q argument in Google's search.
    :return: Search results containing the title, link and snippet of the search result.
    """
    service = build(os.getenv('SERVICE_NAME', 'customsearch'), "v1", developerKey=os.getenv('API_KEY'))
    response = service.cse().list(
        q=search_term,
        cx=os.getenv('ENGINE_ID'),
        cr=os.getenv('COUNTRY_REGION'),
        gl=os.getenv('GEOLOCATION', 'us'),
        lr=os.getenv('RESULT_LANGUAGE', 'lang_en'),
        num=os.getenv('RESULT_NUM', 10),
        fields='items(title,link,snippet)').execute()
    results = response['items']
    __clean_up_snippets(results)

    text_contents = []

    for result in results:
        text_contents.append(
            TextContent(
                type="text",
                text=f"Title: {result['title']}\nLink: {result['link']}\nSnippet: {result['snippet']}"
            )
        )

    return text_contents


def __clean_up_snippets(items: List[dict]) -> None:
    """
    Remove non-breaking space and trailing whitespace from snippets.
    :param items: The search results that contain snippets that have to be cleaned up.
    :return: Nothing, the dict is mutable and updated directly.
    """
    for item in items:
        item.update({k: v.replace('\xa0', ' ').strip() if k == 'snippet' else v for k, v in item.items()})


async def run() -> None:
    options = server.create_initialization_options()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            options
        )