GO Coding Guideline as Agent Skill

 

In this rapidly growing AI age we are using agent to do part of our work (but not all of it yet). The best and stable method for now is an agent integrated into the IDE where we ask for small or medium size tasks.

When I started using it, I found myself constantly asking for changes after the agent had done its work. Many were syntax issues, others were design problems, and some are not following the project standards.

Then, I've decided to put some effort into this, and create a skill. Well, actually I've asked my agent to convert the GO coding guidelines into a skill, and after some minor optimizations, it worked. Now I find myself asking the agent to do the work, and the code generated looks as if I've created it myself. 

I am no longer required.

Create an Agent Skill 

To apply a skill to your agent, create .agents/skills/go-guidelines/SKILL.md

---
name: go-guidelines
description: Apply repository Go standards to implementation, modification, refactoring, and review, including production code, tests, APIs, concurrency, logging, errors, and configuration.
---
# Apply CTO Go guidelines
1. Inspect neighboring code and packages for established patterns.
2. Read only applicable references:
   - [core-style.md](references/core-style.md): always, for changes or reviews.
   - [functions-and-types.md](references/functions-and-types.md): functions, signatures, structs, collections, or packages.
   - [design.md](references/design.md): new components, refactors, dependencies, configuration, or serialization.
   - [errors-and-logging.md](references/errors-and-logging.md): fallible or logging code.
   - [runtime-and-security.md](references/runtime-and-security.md): locks, persistent memory, or backend responses containing secrets.
   - [test.md](references/test.md): unit testing.
3. Apply relevant rules while preserving correctness and public contracts.
4. Treat the references as repository policy. If a rule conflicts with correctness, security, generated-code requirements, or explicit instructions, stop and report it rather than weaken either requirement.
5. In the final response, mention material exceptions and verification performed; do not enumerate rules that were followed normally.

And the create the references files under .agents/skills/go-guidelines/references/

code-style.md

# Core style
## Naming
- Use complete, descriptive words, not abbreviations. Camel-case initialisms: `clientIpAddress`, not `clientIPAddress`.
- Give exported methods and exported struct fields names of at least two words (an IDE workaround).
- Use a one-letter receiver name consistently within a type's methods.
- Export only what another package must access.
## Existing conventions
- Inspect and follow nearby naming, flag, Kafka-topic, package, and construction patterns unless they violate a guideline.
- Do not introduce a competing convention in the same area.
## Comments and cleanup
- Prefer self-explanatory code. Comment only unusual behavior, non-obvious constraints, or required documentation; never narrate obvious code.
- Do not leave TODOs.
- Remove unused variables, functions, constants, comments, commented-out code, and obsolete branches.
- Remove code duplicating Go guarantees.
- Avoid duplicated logic. Extract a shared implementation when it improves clarity and maintains sensible coupling.
## Readability
- Keep nesting to at most three indentation levels. Prefer guard clauses or small cohesive functions.
- Limit function bodies to 15 executable lines. Split by responsibility without increasing coupling or creating trivial wrappers.
- Do not retain a function that only forwards to one simple statement unless it prevents meaningful duplication or provides a real abstraction boundary.

design.md

# Design and dependencies
## Cohesion and coupling
- Minimize parameter and field coupling; keep only shared dependencies and state.
- When functions repeatedly consume the same inputs, create a cohesive object that owns them instead of threading them through every call.
- Prefer object-oriented decomposition for clear ownership, encapsulation, and behavior. Every object must have a cohesive responsibility.
- Keep private implementation details unexported.
## Repository capabilities
- Before implementing common utilities, search and prefer `kit*` packages such as `kiterr`, `kitmap`, `kittime`, and `kitstring`.
- Add reusable, missing general-purpose operations to the appropriate `kit*` package instead of duplicating them locally.
## Configuration and constants
- Do not hard-code operational configuration.
- Put per-PO configuration on the PO and non-PO environment configuration through `core/config.go`.
- Create constants only for values used more than once, not merely to rename one-use literals.
## Serialization
- Add JSON tags only for external communication or Redis persistence, never internal-only structs.

errors-and-logging.md

# Errors and logging
## Error propagation
- Follow `kiterr`'s raise/recover model; do not add errors to ordinary returns.
- Recover or handle errors only at a genuine top-level boundary, such as a new goroutine entry point.
- Never hide an unexpected error. Raise it by default.
- Suppress only explicitly expected errors that retry or recovery cannot resolve; log them at WARN or higher with context.
- Do not conceal caller bugs or invalid state with nil checks, fallbacks, or skipped paths. Surface the defect.
## Diagnostic context
- Include relevant runtime values in raised errors; stack traces identify location, not failing input.
- Never include passwords, secrets, tokens, or other sensitive values in error details or logs.
## Logging
- Use the repository logger, never direct printing.
- Use `pologger` for messages related to a PO.
- Choose verbosity by expected frequency:
  - V1: approximately once per minute.
  - V2: approximately once per second to once per minute.
  - V5: more frequently than once per second.

functions-and-types.md

# Functions and types
## Functions and control flow
- Put every parameter, even a single one, on its own line per repository formatting and nearby syntax.
- Use booleans directly: `if enabled` or `if !enabled`; never compare with `true` or `false`.
- Handle every `switch` default; raise a contextual error if it is impossible or unsupported.
- Minimize function parameters. Do not pass a value separately when it is already available through another parameter.
- Keep shared method dependencies and state in receiver fields instead of repeatedly passing them.
## Structs and packages
- Define named struct types. Do not use anonymous inline structs.
- Keep one principal class-like component per package. Follow existing layout for related helpers and types; do not harm cohesion by mechanically creating packages.
- Pass and return struct pointers, including `time.Time`, unless a required interface or established immutable-value contract dictates otherwise.
- Store struct elements in slices as pointers: `[]*Item`, not `[]Item`. This concerns elements, not pointers to slices.
- Represent empty slices as `nil` unless an external serialization contract requires `[]`.
- Replace deeply nested or otherwise complex composite types with named structs that express each layer.

runtime-and-security.md

# Runtime safety and security
## Persistent memory
- Bound every persistent in-memory collection or cache.
- Use eviction, expiration, or another limit to prevent sustained traffic exhausting pod memory.
## Backend secrets
- Never return passwords from backend APIs to the frontend.
- Exclude credentials and equivalent secrets from response models, serialization, logs, and error details.
## Locking
- Guarantee lock cleanup on every exit, including raised errors.
- Prefer a small function that acquires the lock and immediately defers its unlock before protected work.
- Minimize critical sections; avoid unknown or blocking calls while locked when possible.

test.md

# Verification
- do not create unit test in the same folder as the tested source, tests live in `images/tests/src/project/tests`.
- Follow existing test conventions.
- Prefer end-to-end simulation over testing only the updated file.

Final Note

The nice thing about this skill is that is is split to multiple sections, so first the skill is loaded only when required, and second only the relevant sub skill items are loaded, so we save tokens and optimize the agent functionality.

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

WOWOWOW BREAKING CHANGE:
pip3.12 install mcp fastapi uvicorn pyjwt cryptography httpx  authlib python-dotenv itsdangerous pyjwt 

MCP updated to version 2.0, this works only with MCP 1.x, so use:

pip3.12 install "mcp>=1.28,<2" 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 

OAuth implementation With Refresh Token

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")


# How long an MCP access token lives before the client must refresh it.
ACCESS_TOKEN_TTL = 3600
# How long an MCP refresh token lives before it must be re-issued via a fresh
# /authorize + Google login. None of this is tied to Google's own tokens: we
# only ever use Google to establish who the human is, then mint our own
# short-lived access token / long-lived refresh token pair for the MCP client.
REFRESH_TOKEN_TTL = 60 * 60 * 24 * 30  # 30 days


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] = {}
        self.mcp_refresh_tokens: dict[str, RefreshToken] = {}
        # keep access <-> refresh pairs linked so we can rotate/revoke both
        # halves together without scanning every dict.
        self._access_to_refresh: dict[str, str] = {}
        self._refresh_to_access: dict[str, str] = {}
        # 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 = self._issue_token_pair(client.client_id, authorization_code.scopes)
        del self.mcp_auth_codes[authorization_code.code]
        return token

    def _issue_token_pair(self, client_id: str, scopes: list[str]) -> OAuthToken:
        """Mint a fresh access token + refresh token pair and register the
        bookkeeping needed to look them up / rotate / revoke them later."""
        access_token = secrets.token_urlsafe(32)
        refresh_token = secrets.token_urlsafe(32)
        now = int(time.time())

        self.mcp_tokens[access_token] = AccessToken(
            token=access_token,
            client_id=client_id,
            scopes=scopes,
            expires_at=now + ACCESS_TOKEN_TTL,
        )
        self.mcp_refresh_tokens[refresh_token] = RefreshToken(
            token=refresh_token,
            client_id=client_id,
            scopes=scopes,
            expires_at=now + REFRESH_TOKEN_TTL,
        )
        self._access_to_refresh[access_token] = refresh_token
        self._refresh_to_access[refresh_token] = access_token

        return OAuthToken(
            access_token=access_token,
            token_type="bearer",
            expires_in=ACCESS_TOKEN_TTL,
            scope=" ".join(scopes),
            refresh_token=refresh_token,
        )

    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():
            self._discard_access(token)
            return None
        return access

    async def load_refresh_token(self, client, refresh_token: str) -> RefreshToken | None:
        rt = self.mcp_refresh_tokens.get(refresh_token)
        if rt is None or rt.client_id != client.client_id:
            return None
        if rt.expires_at and rt.expires_at < time.time():
            self._discard_refresh(refresh_token)
            return None
        return rt

    async def exchange_refresh_token(
        self,
        client: OAuthClientInformationFull,
        refresh_token: RefreshToken,
        scopes: list[str],
    ) -> OAuthToken:
        # Requested scopes must not exceed what the original grant allowed.
        if scopes:
            if not set(scopes).issubset(set(refresh_token.scopes)):
                raise ValueError("Requested scopes exceed the original grant")
            granted_scopes = scopes
        else:
            granted_scopes = refresh_token.scopes

        # Rotate: the old access token (if still around) and the refresh
        # token being redeemed both get invalidated, and a brand-new pair
        # takes their place. This limits the blast radius if a refresh
        # token is ever replayed by an attacker.
        self._discard_refresh(refresh_token.token)

        return self._issue_token_pair(client.client_id, granted_scopes)

    def _discard_access(self, access_token: str) -> None:
        self.mcp_tokens.pop(access_token, None)
        rt = self._access_to_refresh.pop(access_token, None)
        if rt:
            self._refresh_to_access.pop(rt, None)

    def _discard_refresh(self, refresh_token: str) -> None:
        self.mcp_refresh_tokens.pop(refresh_token, None)
        at = self._refresh_to_access.pop(refresh_token, None)
        if at:
            self.mcp_tokens.pop(at, None)
            self._access_to_refresh.pop(at, None)

    async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
        # The SDK hands us the token object itself (not the raw string), and
        # it may be either an access or a refresh token depending on what the
        # client asked to revoke. Discard whichever side it is; both helpers
        # already clean up the paired token too.
        if isinstance(token, RefreshToken):
            self._discard_refresh(token.token)
        else:
            self._discard_access(token.token)


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.