Skip to content

Web Search Integration with RequestCompletion

This guide demonstrates how to create a web search integration for RequestCompletion (RC) using both MCP (Model Context Protocol) servers and custom Google API tools. This setup allows your AI agent to search the web and fetch content from URLs.

Prerequisites

Before implementing this integration, you'll need:

  1. Google Custom Search API credentials:
  2. Visit the Google Cloud Console
  3. Enable the Custom Search API

  4. Create API credentials and a Custom Search Engine ID

  5. Environment variables: 

    GOOGLE_SEARCH_API_KEY=your_api_key_here
GOOGLE_SEARCH_ENGINE_ID=your_search_engine_id_here

  6. Required packages: 

    pip install railtracks python-dotenv aiohttp

Implementation

Step 1: Import Dependencies and Load Environment

from dotenv import load_dotenv
import os
from railtracks import connect_mcp
import railtracks as rt
import asyncio
from railtracks.rt_mcp import MCPHttpParams
import aiohttp
from typing import Dict, Any

load_dotenv()

Step 2: Set Up MCP Tools for URL Fetching

The MCP server provides tools that can fetch and process content from URLs:

# MCP Tools that can fetch data from URLs
fetch_mcp_server = connect_mcp(MCPHttpParams(url="https://remote.mcpservers.org/fetch/mcp"))
fetch_mcp_tools = fetch_mcp_server.tools
Read more about the from_mcp_server utility TODO: change this link.
This connects to a remote MCP server that provides URL fetching capabilities.

Step 3: Create Custom Google Search Tool

def _format_results(data: Dict[str, Any]) -> Dict[str, Any]:
    return data


@rt.function_node
async def google_search(query: str, num_results: int = 3) -> Dict[str, Any]:
   """
   Tool for searching using Google Custom Search API

   Args:
       query (str): The search query
       num_results (int): The number of results to return (max 5)

   Returns:
       Dict[str, Any]: Formatted search results
   """
   params = {
      'key': os.environ['GOOGLE_SEARCH_API_KEY'],
      'cx': os.environ['GOOGLE_SEARCH_ENGINE_ID'],
      'q': query,
      'num': min(num_results, 5)  # Google API maximum is 5
   }

   async with aiohttp.ClientSession() as session:
      try:
         async with session.get("https://www.googleapis.com/customsearch/v1", params=params) as response:
            if response.status == 200:
               data = await response.json()
               return _format_results(data)
            else:
               error_text = await response.text()
               raise Exception(f"Google API error {response.status}: {error_text}")
      except Exception as e:
         raise Exception(f"Search failed: {str(e)}")

Step 4: Create and Use the Search Agent

# Combine all tools
tools = fetch_mcp_tools + [google_search]

# Create the agent with search capabilities
WebSearchAgent = rt.agent_node(
    # tool_nodes={*tools},    # Uncomment this line to use the tools
    system_message="""You are an information gathering agent that can search the web.""",
    llm=rt.llm.OpenAILLM("gpt-4o"),
)

# Example usage
user_prompt = """Tell me about Railtown AI."""
async def call_web_search():
    result = await rt.call(agent, message_history)
    print(result)

# asyncio.run(call_node())

How It Works

  1. Google Search Tool: Uses the Google Custom Search API to find relevant web pages based on user queries
  2. MCP Fetch Tools: Retrieves and processes content from the URLs found in search results
  3. Agent Integration: Combines both tools to create a comprehensive web search and content analysis system