web_search.py

import os
import asyncio
import httpx
from tavily import TavilyClient
from mcp.server.fastmcp import FastMCP
import nest_asyncio
from dotenv import load_dotenv

# Apply nested asyncio patch
nest_asyncio.apply()

GEMINI_API_KEY="AIzaSyCUCivstFpC9pq_jMHMYdlPrmh9Bx97dFo"
TAVILY_API_KEY="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"
OPENAI_API_KEY="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"
QDRANT_API_KEY="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2Nlc3MiOiJtIiwiZXhwIjoxNzUxMDUxNzg4fQ.I9J-K7OM0BtcNKgj2d4uVM8QYAHYfFCVAyP4rlZkK2E"
QDRANT_URL="https://6a3aade6-e8ad-4a6c-a579-21f5af90b7e8.us-east4-0.gcp.cloud.qdrant.io"
OPENAI_API_KEY="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"
WEAVIATE_URL="https://xbvlj5rpqyiswspww0tthq.c0.us-west3.gcp.weaviate.cloud"
WEAVIATE_API_KEY="RU9acU1CYnNRTjY1S1ZFc18zNS9tQktaWlcwTzFEUjlscEVCUGF4YU5xRWx2MDhmTUtIdUhnOWdOTGVZPV92MjAw"
DEEPINFRA_API_KEY="285LUJulGIprqT6hcPhiXtcrphU04FG4"
DEEPINFRA_BASE_URL="https://api.deepinfra.com/v1/openai"


# Initialize Tavily client and FastMCP
tavily_client = TavilyClient(api_key=TAVILY_API_KEY)
mcp = FastMCP("tavily_search")

# Define the search tool
@mcp.tool()
async def search_autism(query: str) -> dict:
    """Performs a Tavily web search for information about autism."""
    try:
        # Execute a search query
        response = tavily_client.search(
            query=query,
            max_results=5,
            search_depth="advanced",
            topic="general",
            name="live_search",
            description="""Using the latest information from reputable online sources, provide a concise AI-generated overview of the query related to autism spectrum  """,
            include_answer=True
        )
        return {
            "results": response.get("results", []),
            "answer": response.get("answer", "")
        }
    except Exception as e:
        return {"error": f"Search failed: {str(e)}"}

# Main entry point
# async def main():
#     query = "autism symptoms and treatments"  # Replace with dynamic input if needed
#     result = await search_autism(query)
#     print("Search Results:")
#     for res in result.get("results", []):
#         print(f"- {res.get('title')} ({res.get('url')})")
#     print("\nAnswer:")
#     print(result.get("answer", "No answer provided."))

# Run the script
if __name__ == "__main__":
    asyncio.run(main())