Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, Depends, HTTPException, Header | |
| from typing import Optional | |
| try: | |
| from .models import SearchRequest, SearchResponse | |
| from .github_client import GitHubClient | |
| except ImportError: | |
| from models import SearchRequest, SearchResponse | |
| from github_client import GitHubClient | |
| import os | |
| app = FastAPI(title="MCP GitHub PR Opportunity Server") | |
| def get_github_token(authorization: Optional[str] = Header(None)) -> str: | |
| if authorization and authorization.startswith("token "): | |
| return authorization.split(" ", 1)[1] | |
| token = os.getenv("GITHUB_TOKEN") | |
| if not token: | |
| raise HTTPException(status_code=401, detail="GitHub token required in Authorization header or .env") | |
| return token | |
| def search_pr_opportunities( | |
| request: SearchRequest, | |
| token: str = Depends(get_github_token) | |
| ): | |
| try: | |
| client = GitHubClient(token) | |
| opportunities, total_count = client.find_pr_opportunities( | |
| keyword=request.keyword, | |
| topic=request.topic, | |
| per_page=request.per_page, | |
| page=request.page | |
| ) | |
| return SearchResponse(opportunities=opportunities, total_count=total_count) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) |