Spaces:
Sleeping
Sleeping
File size: 2,696 Bytes
0794bda e427195 0794bda | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """
HTTP wrapper for MCP server deployment on Hugging Face Spaces.
"""
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
import uvicorn
import os
from dotenv import load_dotenv
from server import mcp
# Load environment variables from .env file
load_dotenv()
# Create FastAPI app
app = FastAPI(
title="Basicsearch MCP Server",
description="YouTube video search MCP server"
)
@app.get("/")
async def root():
"""Health check endpoint"""
return {
"status": "healthy",
"service": "basicsearch-mcp-server",
"tools": ["search_youtube_videos"]
}
@app.get("/health")
async def health():
"""Health check endpoint"""
return {"status": "ok"}
@app.get("/debug/env")
async def debug_env():
"""Debug endpoint to check if YOUTUBE_API_KEY is loaded"""
api_key = os.getenv('YOUTUBE_API_KEY')
return {
"youtube_api_key_exists": api_key is not None,
"youtube_api_key_length": len(api_key) if api_key else 0,
"youtube_api_key_prefix": api_key[:10] + "..." if api_key else "NOT SET",
"all_env_vars": [key for key in os.environ.keys() if 'YOUTUBE' in key or 'API' in key]
}
@app.post("/sse")
async def sse_endpoint(request: Request):
"""SSE endpoint for MCP protocol"""
# This endpoint would handle SSE connections for MCP
# For now, return a basic response
return {"message": "SSE endpoint for MCP protocol"}
@app.get("/tools")
async def list_tools():
"""List available tools"""
return {
"tools": [
{
"name": "search_youtube_videos",
"description": "Searches YouTube for videos based on a query.",
"parameters": {
"query": "string (required)",
"max_results": "integer (optional, default: 5)"
}
}
]
}
@app.post("/search")
async def search_youtube(request: Request):
"""Direct HTTP endpoint for YouTube search"""
from server import search_youtube_videos
data = await request.json()
query = data.get("query")
max_results = data.get("max_results", 5)
if not query:
return JSONResponse(
status_code=400,
content={"error": "Query parameter is required"}
)
try:
results = search_youtube_videos(query, max_results)
return {"results": results}
except Exception as e:
return JSONResponse(
status_code=500,
content={"error": str(e)}
)
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
uvicorn.run(app, host="0.0.0.0", port=port)
|