Spaces:
Sleeping
Sleeping
File size: 2,092 Bytes
76d3296 b93b962 76d3296 b93b962 76d3296 b93b962 76d3296 b93b962 |
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 |
"""
Run from the repository root:
uv run examples/snippets/servers/oauth_server.py
"""
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
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.responses import JSONResponse
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"],
),
port=7860
)
@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%",
}
# Create a combined lifespan to manage both session managers
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
async with contextlib.AsyncExitStack() as stack:
await stack.enter_async_context(mcp.session_manager.run())
yield
async def homepage(request):
return JSONResponse({'hello': 'world'})
# Create the Starlette app and mount the MCP servers
app = Starlette(
routes=[
Mount("/sub", mcp.streamable_http_app()),
Route("/", homepage),
],
lifespan=lifespan,
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |