Spaces:
Running
on
CPU Upgrade
Running
on
CPU Upgrade
File size: 3,306 Bytes
76d3296 439a81e 312d184 80a2cf1 87d58b2 76d3296 312d184 76d3296 b93b962 80a2cf1 b93b962 80a2cf1 b93b962 312d184 b93b962 86a2dd4 d7dfab2 86a2dd4 76d3296 b93b962 86a2dd4 |
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 97 98 99 100 101 102 103 104 |
from pydantic import AnyHttpUrl
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, Request
from starlette.responses import JSONResponse, RedirectResponse
import contextlib
import httpx
class SimpleTokenVerifier(TokenVerifier):
"""Simple token verifier for demonstration."""
async def verify_token(self, token: str) -> AccessToken | None:
pass # This is where you would implement actual token validation
# Create FastMCP instance as a Resource Server
mcp = FastMCP(
"Weather Service",
# Token verifier for authentication
token_verifier=SimpleTokenVerifier(),
# Auth settings for RFC 9728 Protected Resource Metadata
auth=AuthSettings(
issuer_url=AnyHttpUrl("https://huggingface.co/.well-known/oauth-authorization-server"), # Authorization Server URL
resource_server_url=AnyHttpUrl("https://freddyaboulton-fastmcp-oauth.hf.space/"), # This server's URL
required_scopes=["user"],
),
)
@mcp.tool()
async def get_weather(city: str = "London") -> dict[str, str]:
"""Get weather data for a city"""
return {
"city": city,
"temperature": "22",
"condition": "Partly cloudy",
"humidity": "65%",
}
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
async with contextlib.AsyncExitStack() as stack:
await stack.enter_async_context(mcp.session_manager.run())
yield
app = FastAPI(lifespan=lifespan)
app.mount("/weather_mcp", mcp.streamable_http_app())
@app.get("/")
async def _():
return JSONResponse({"message": "Welcome to the Weather Service!"})
@app.get("/.well-known/oauth-protected-resource")
async def _():
return RedirectResponse("/weather_mcp/.well-known/oauth-protected-resource")
@app.post("/register")
async def _():
return RedirectResponse("https://huggingface.co/oauth/register")
@app.post("/register")
async def register_proxy(request: Request):
"""Forward requests to Hugging Face OAuth register endpoint"""
target_url = "https://huggingface.co/oauth/register"
async with httpx.AsyncClient() as client:
# Prepare the request data
headers = dict(request.headers)
# Remove host header to avoid conflicts
headers.pop("host", None)
# Get query parameters
query_params = str(request.url.query) if request.url.query else None
target_url_with_params = f"{target_url}?{query_params}" if query_params else target_url
# Get request body if present
body = await request.body()
print("body", body)
# Forward the request
response = await client.request(
method=request.method,
url=target_url_with_params,
headers=headers,
content=body,
follow_redirects=False
)
# Return the response with the same status code, headers, and content
return Response(
content=response.content,
status_code=response.status_code,
headers=dict(response.headers)
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
|