Setup EC2 MCP Server with oAuth

In this post we list the steps required to setup MCP server with Google oAuth on a EC2 server.

We use the standard method for authentication, which means that MCP clients like Claude and VSCode can used this. 

Notice this implemention does not use Dynamic Client Registration, so we will manually get the client ID from the server.

I've spent a very long time on the python code due to some bugs and a very poor documentation on FastMCP.

It make the implementation not easy, be prepared!

1. Prepare The Server

Create an EC2 Instance

• instance type: t3.medium
• Open ingress ports: 80, 443, 22
• Assign elastic IP to the EC2 instance

Add DNS Entry

Create a new entry in you DNS to point to the server. 

In this case we will use the DNS name: mcp-auth.alonana.com

The DNS will be later used to create SSL certificate.

2. Simple MCP Server Update

Login to the EC2

ssh ec2-user@mcp-auth.alonana.com

Install Docker

sudo dnf update -y

sudo dnf install -y docker git
sudo systemctl enable docker
sudo systemctl start docker
sudo usermod -aG docker ec2-user
newgrp docker

Install Python

sudo dnf install python3.12 python3.12-pip -y
python3.12 -m pip install --upgrade pip
python3.12 -m venv venv 
source ~/venv/bin/activate
pip3.12 install mcp fastapi uvicorn pyjwt cryptography httpx  authlib python-dotenv itsdangerous pyjwt 

Basic Server Code

Create app.py

from fastapi import FastAPI
app = FastAPI(
    title="Example MCP Server"
)

@app.get("/health")
def health():
    return {
        "status": "ok"
    }

@app.get("/")
def root():
    return {
        "service": "mcp-server"
    }

Uvicorn as Service

sudo vi /etc/systemd/system/mcp-server.service

[Unit]
Description=MCP FastAPI Server
After=network.target
[Service]
User=ec2-user
Group=ec2-user
WorkingDirectory=/home/ec2-user
Environment="PATH=/home/ec2-user/venv/bin"

ExecStart=/home/ec2-user/venv/bin/uvicorn app:app \

    --host 127.0.0.1 \

    --port 8080 \

    --proxy-headers \

    --forwarded-allow-ips="*"

Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target

Then run as service

sudo systemctl daemon-reload
sudo systemctl enable mcp-server
sudo systemctl start mcp-server

Check status and logs

sudo systemctl --no-pager status mcp-server

sudo journalctl --no-pager -u mcp-server 

Nginx

Install

sudo dnf install nginx -y
sudo systemctl enable nginx
sudo systemctl start nginx

sudo vi /etc/nginx/conf.d/mcp.conf

server {

    listen 80;
    server_name mcp-auth.alonana.com;

    location / {
        proxy_pass http://127.0.0.1:8080;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;

        proxy_set_header X-Forwarded-Proto https;

        proxy_set_header X-Forwarded-Host $host;

        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        proxy_set_header Authorization $http_authorization;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Apply to NGINX

sudo systemctl reload nginx

TLS with Let's Encrypt

Install

sudo dnf install certbot python3-certbot-nginx -y

Request certificate

sudo certbot --nginx -d mcp-auth.alonana.com

check TLS is working and ports are allowed, run from your laptop

curl https://mcp-auth.alonana.com/health

3. OAuth MCP Server Update

Google OAuth Creation

Open Google console. 

In the search bar type: OAuth consent screen

Create new Project.

Select the new project.

Click Get Started.

Make sure to select "External" as the type.

In the Branding tab, authorized domains, add only the domain name without the prefix:

alonana.com

In Audiance tab, under Test users, add your gmail email address.

In Clients tab, add new client

Type: Web application.

Authorized URLs, add the URLs:

http://localhost:8000/auth/callback

https://mcp-auth.alonana.com/auth/callback

https://mcp-auth.alonana.com/oauth/google/callback

copy the client ID and secrets, and save them in the EC2

vi .env

GOOGLE_CLIENT_ID=313236448176-o06pa60m4hok8mn58ckpqugm8q4d6m7v.apps.googleusercontent.com
GOOGLE_CLIENT_SECRET=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

BASE_URL=https://mcp-auth.alonana.com

SESSION_SECRET=generate-a-long-random-secret-gkjhf-gfd-gfd-tre43

Update the service

sudo vi /etc/systemd/system/mcp-server.service

add after the WorkingDirectoty

EnvironmentFile=/home/ec2-user/.env

Reload the service

sudo systemctl daemon-reload
sudo systemctl restart mcp-server

sudo systemctl --no-pager status mcp-server

sudo journalctl --no-pager -u mcp-server 

OAuth implementation

Wow, I mean that was very difficult.

We have FixIssuerTrailingSlashMiddleware as a bypass to FastMCP bug.

We have GoogleAuthServerProvider as a our own wrapping over Google OAuth.

We have TransportSecuritySettings to overcome blocking hosts redirection.

We have "app.mount" with empty argument to enable mcp with oAuth.

Each of this was hard to find and documentation is very bad.

vi app.py

import os
import json
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse

from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

from authlib.integrations.starlette_client import OAuth
from dotenv import load_dotenv

from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

import secrets
import time
import httpx
from mcp.server.auth.provider import (
    OAuthAuthorizationServerProvider,
    AuthorizationParams,
    AuthorizationCode,
    AccessToken,
    RefreshToken,
)
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions



WELL_KNOWN_PATHS = {
    "/.well-known/oauth-authorization-server",
    "/.well-known/oauth-protected-resource",
}

class FixIssuerTrailingSlashMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        if request.url.path not in WELL_KNOWN_PATHS:
            return response

        body = b""
        async for chunk in response.body_iterator:
            body += chunk

        try:
            data = json.loads(body)
            if isinstance(data.get("issuer"), str):
                data["issuer"] = data["issuer"].rstrip("/")
            if isinstance(data.get("resource"), str):
                data["resource"] = data["resource"].rstrip("/")
            if isinstance(data.get("authorization_servers"), list):
                data["authorization_servers"] = [u.rstrip("/") for u in data["authorization_servers"]]
            body = json.dumps(data).encode()
        except (json.JSONDecodeError, AttributeError):
            pass

        headers = {k: v for k, v in response.headers.items() if k.lower() != "content-length"}
        return Response(content=body, status_code=response.status_code, headers=headers, media_type="application/json")




class GoogleAuthServerProvider(OAuthAuthorizationServerProvider):
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url  # e.g. https://mcp-auth.alonana.com

        self.clients: dict[str, OAuthClientInformationFull] = {}
        self.mcp_auth_codes: dict[str, AuthorizationCode] = {}
        self.mcp_tokens: dict[str, AccessToken] = {}
        # correlates our internal "state" with the original MCP request
        self.pending: dict[str, dict] = {}

    # --- Dynamic Client Registration: MCP clients self-register here ---
    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
        return self.clients.get(client_id)

    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
        self.clients[client_info.client_id] = client_info

    # --- Step 1: MCP client hits /authorize, we bounce to Google ---
    async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
        state = secrets.token_urlsafe(32)
        self.pending[state] = {
            "client_id": client.client_id,
            "redirect_uri": str(params.redirect_uri),
            "code_challenge": params.code_challenge,
            "scopes": params.scopes or ["mcp:tools"],
            "mcp_state": params.state,
        }
        google_params = {
            "client_id": self.client_id,
            "redirect_uri": f"{self.base_url}/oauth/google/callback",
            "response_type": "code",
            "scope": "openid email profile",
            "state": state,
            "access_type": "online",
            "prompt": "select_account",
        }
        return f"{GOOGLE_AUTH_URL}?{httpx.QueryParams(google_params)}"

    # --- Step 3: MCP client exchanges our code for an MCP token ---
    async def load_authorization_code(self, client, authorization_code: str) -> AuthorizationCode | None:
        return self.mcp_auth_codes.get(authorization_code)

    async def exchange_authorization_code(self, client, authorization_code: AuthorizationCode) -> OAuthToken:
        token = secrets.token_urlsafe(32)
        self.mcp_tokens[token] = AccessToken(
            token=token,
            client_id=client.client_id,
            scopes=authorization_code.scopes,
            expires_at=int(time.time()) + 3600,
        )
        del self.mcp_auth_codes[authorization_code.code]
        return OAuthToken(access_token=token, token_type="bearer", expires_in=3600,
                           scope=" ".join(authorization_code.scopes))

    async def load_access_token(self, token: str) -> AccessToken | None:
        access = self.mcp_tokens.get(token)
        if access and access.expires_at and access.expires_at < time.time():
            del self.mcp_tokens[token]
            return None
        return access

    async def load_refresh_token(self, client, refresh_token: str) -> RefreshToken | None:
        return None  # keep it simple: no refresh tokens for now

    async def exchange_refresh_token(self, client, refresh_token, scopes):
        raise NotImplementedError

    async def revoke_token(self, token) -> None:
        self.mcp_tokens.pop(token, None)


load_dotenv()

auth_provider = GoogleAuthServerProvider(
    client_id=os.environ["GOOGLE_CLIENT_ID"],
    client_secret=os.environ["GOOGLE_CLIENT_SECRET"],
    base_url="https://mcp-auth.alonana.com",
)

# =====================================================
# MCP SERVER
# =====================================================

mcp = FastMCP(
    "example-server",
    auth_server_provider=auth_provider,
    auth=AuthSettings(
        issuer_url="https://mcp-auth.alonana.com",
        resource_server_url="https://mcp-auth.alonana.com",
        client_registration_options=ClientRegistrationOptions(
            enabled=True,
            valid_scopes=["mcp:tools"],
            default_scopes=["mcp:tools"],
        ),
    ),
    transport_security=TransportSecuritySettings(
        allowed_hosts=["mcp-auth.alonana.com", "localhost:*", "127.0.0.1:*"],
        allowed_origins=["https://mcp-auth.alonana.com"],
    ),
)
GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"





@mcp.tool()
def hello(name: str) -> str:
    """
    Simple greeting tool.
    """
    return f"Hello {name}"


@mcp.tool()
def get_status() -> str:
    """
    Health check tool.
    """
    return "MCP server is running"


# =====================================================
# FASTAPI APPLICATION
# =====================================================

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with mcp.session_manager.run():
        yield


app = FastAPI(
    title="Example MCP Server",
    lifespan=lifespan,
)


# =====================================================
# TRUSTED HOSTS
# =====================================================

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=[
        "mcp-auth.alonana.com",
        "localhost",
        "127.0.0.1",
    ],
)


# =====================================================
# SESSION
# =====================================================

app.add_middleware(
    SessionMiddleware,
    secret_key=os.environ["SESSION_SECRET"],
)


# =====================================================
# GOOGLE OAUTH
# =====================================================

oauth = OAuth()


oauth.register(
    name="google",
    client_id=os.environ["GOOGLE_CLIENT_ID"],
    client_secret=os.environ["GOOGLE_CLIENT_SECRET"],
    server_metadata_url=(
        "https://accounts.google.com/.well-known/openid-configuration"
    ),
    client_kwargs={
        "scope": "openid email profile",
    },
)


# =====================================================
# NORMAL ENDPOINTS
# =====================================================

@app.get("/")
async def root(request: Request):

    user = request.session.get("user")

    return {
        "service": "mcp-server",
        "logged_in": user is not None,
        "user": user,
        "mcp_endpoint": "/mcp",
    }


@app.get("/health")
async def health():

    return {
        "status": "ok"
    }


# =====================================================
# HUMAN AUTH
# =====================================================

@app.get("/login")
async def login(request: Request):

    redirect_uri = request.url_for(
        "auth_callback"
    )

    return await oauth.google.authorize_redirect(
        request,
        redirect_uri,
    )


@app.get("/auth/callback")
async def auth_callback(request: Request):

    token = await oauth.google.authorize_access_token(
        request
    )

    user = token["userinfo"]

    request.session["user"] = {
        "id": user.get("sub"),
        "name": user.get("name"),
        "email": user.get("email"),
        "picture": user.get("picture"),
    }

    return RedirectResponse(
        url="/"
    )


@app.get("/me")
async def me(request: Request):

    user = request.session.get("user")

    if not user:
        return {
            "authenticated": False
        }

    return {
        "authenticated": True,
        "user": user,
    }


@app.get("/logout")
async def logout(request: Request):

    request.session.clear()

    return {
        "message": "Logged out"
    }


# =====================================================
# MCP CLIENT AUTH
# =====================================================

@app.get("/oauth/google/callback")
async def google_oauth_callback(request: Request):
    state = request.query_params["state"]
    code = request.query_params["code"]

    pending = auth_provider.pending.pop(state, None)
    if not pending:
        return {"error": "invalid or expired state"}

    async with httpx.AsyncClient() as client:
        resp = await client.post(GOOGLE_TOKEN_URL, data={
            "code": code,
            "client_id": auth_provider.client_id,
            "client_secret": auth_provider.client_secret,
            "redirect_uri": f"{auth_provider.base_url}/oauth/google/callback",
            "grant_type": "authorization_code",
        })
    resp.raise_for_status()

    mcp_code = secrets.token_urlsafe(32)
    auth_provider.mcp_auth_codes[mcp_code] = AuthorizationCode(
        code=mcp_code,
        client_id=pending["client_id"],
        scopes=pending["scopes"],
        expires_at=int(time.time()) + 300,
        code_challenge=pending["code_challenge"],
        redirect_uri=pending["redirect_uri"],
        redirect_uri_provided_explicitly=True,
    )

    redirect_url = f"{pending['redirect_uri']}?code={mcp_code}&state={pending['mcp_state']}"
    return RedirectResponse(url=redirect_url)


# =====================================================
# MCP HTTP ENDPOINT
# =====================================================


app.add_middleware(FixIssuerTrailingSlashMiddleware)

# MCP SDK 1.28.1 already exposes /mcp internally.
# Therefore mount at root.
app.mount(
    "",
    mcp.streamable_http_app()
)


# =====================================================
# LOCAL RUN
# =====================================================

if __name__ == "__main__":

    import uvicorn

    uvicorn.run(
        "app:app",
        host="0.0.0.0",
        port=8080,
        reload=True,
    )

Reload the service

sudo systemctl daemon-reload
sudo systemctl restart mcp-server

sudo systemctl status mcp-server

sudo journalctl -u mcp-server 

Using the MCP

Get a client ID from the server manually:

curl -s -X POST https://mcp-auth.alonana.com/register \
  -H 'Content-Type: application/json' \
  -d '{
    "redirect_uris": ["http://portal/oauth-callback.html"],
    "grant_types": ["authorization_code", "refresh_token"],
    "response_types": ["code"],
    "token_endpoint_auth_method": "none"
  }' 

Use this client ID for new connection creation

Final Note

In case you survived until this point, well done!

Notice that in case of problems, do not use ChatGPT, as it is not smart enough to understand the issues. I've used claude AI which provided great assistance.