Working Example (Python)

This guide builds upon the official MCP Server tutorial to create a remote MCP server with AgentPay integration. We’ll start with a basic weather server and enhance it with remote capabilities and monetization.

Prerequisites

Python 3.10 or higher
Basic understanding of MCP (from the official tutorial)
An AgentPay account and Service Token (see Quickstart Guide)

AgentPay is currently in Early Access. To get early access to AgentPay, please sign up for the Waitlist here.

Step 1: Project Setup

First, create a new project directory and set up your environment:

# Create project directory
mkdir weather-server
cd weather-server

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install "mcp[cli]" starlette uvicorn httpx python-dotenv agentpay-sdk

The agentpay-sdk package currently on PyPI is a placeholder to reserve the name during Early Access. To get the actual SDK now, join the Waitlist.

Create a .env file to store your AgentPay Service Token:

# .env
AGENTPAY_SERVICE_TOKEN=your_service_token_here

Step 2: Basic Server Implementation

Create weather_server.py with the basic MCP server structure per the official MCP Server tutorial:

import os
from typing import Any, Dict, Optional
import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Initialize FastMCP
mcp = FastMCP("weather-server")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

# Helper function for NWS API requests
async def make_nws_request(url: str) -> Optional[Dict[str, Any]]:
    """Make a request to the NWS API with proper error handling."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

# MCP Tools
@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    if not state or len(state) != 2:
        return "Error: Please provide a valid two-letter US state code."

    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "No active alerts for this state."

    alerts = []
    for feature in data["features"]:
        props = feature["properties"]
        alert = f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
"""
        alerts.append(alert)

    return "\n---\n".join(alerts)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data or "properties" not in points_data:
        return "Error: Unable to fetch forecast data for this location."

    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data or "properties" not in forecast_data:
        return "Error: Unable to fetch detailed forecast."

    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Show next 5 periods
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

if __name__ == "__main__":
    mcp.run(transport='stdio')

Test the basic server:

python weather_server.py

Step 3: Add Remote Server Capabilities

Now, let’s modify the server to run as a remote HTTP server using Starlette. Update weather_server.py to add the necessary imports and server setup:

from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
import uvicorn

# Create Starlette app with FastMCP mounted at root (add this before the if __name__ == "__main__" block)
app = Starlette(
    routes=[
        # Mount the FastMCP SSE app at the root to handle MCP protocol
        Mount("/", app=mcp.sse_app())
    ],
    middleware=[
        # Enable CORS for development (customize for production)
        Middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )
    ]
)

# Update the main block
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

This setup:

Mounts the FastMCP SSE app at the root path to handle MCP protocol
Enables CORS for development (you should customize this for production)
Uses uvicorn to run the server

Test the remote server:

python weather_server.py

Your server is now running at http://localhost:8000 and ready to accept requests from MCP clients.

Step 4: Add AgentPay Integration

Now, let’s integrate AgentPay following our four key steps:

i. Initialize AgentPayClient

Add the AgentPay client initialization near the top of the file:

from agentpay_sdk import AgentPayClient

# Add after FastMCP initialization
agentpay_client = AgentPayClient(service_token=os.getenv("AGENTPAY_SERVICE_TOKEN"))

ii. Extract User API Key

Add the context variable and middleware to extract the API key from the X-AGENTPAY-API-KEY header:

from contextvars import ContextVar
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

# Add context variable after AgentPay client initialization
api_key_context: ContextVar[str | None] = ContextVar("api_key_context", default=None)

# Add the API Key middleware class
class ApiKeyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract API key from header
        api_key = request.headers.get("X-AGENTPAY-API-KEY")
        
        # Set API key in context for use in route handlers
        token = api_key_context.set(api_key)
        try:
            response = await call_next(request)
        finally:
            api_key_context.reset(token)
        return response

# Update the Starlette app to include the middleware
app = Starlette(
    routes=[Mount("/", app=mcp.sse_app())],
    middleware=[
        Middleware(CORSMiddleware, ...),  # Previous CORS middleware
        Middleware(ApiKeyMiddleware)      # Add API key middleware
    ]
)

iii. Validate User API Key

Update the middleware to validate the API key:

class ApiKeyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        api_key = request.headers.get("X-AGENTPAY-API-KEY")

        if api_key:
            try:
                # Validate API key immediately after extraction
                validation_result = agentpay_client.validate_api_key(api_key=api_key)
                if not validation_result.is_valid:
                    return JSONResponse(
                        {"error": "Unauthorized", "message": f"Invalid API Key: {validation_result.invalid_reason}"},
                        status_code=401
                    )
            except Exception as e:
                return JSONResponse(
                    {"error": "Internal Server Error", "message": "API Key validation failed"},
                    status_code=500
                )

        # Set API key in context if valid
        token = api_key_context.set(api_key)
        try:
            response = await call_next(request)
        finally:
            api_key_context.reset(token)
        return response

iv. Consume Usage

Finally, update the MCP tools to charge for usage:

import uuid

# Add these constants
ALERT_COST_CENTS = 2  # 2 cents per alert check
FORECAST_COST_CENTS = 3  # 3 cents per forecast

# Update get_alerts tool
@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    # Get API key from context
    api_key = api_key_context.get()
    if not api_key:
        return "Error: API Key missing"

    # Charge for usage
    usage_id = str(uuid.uuid4())
    result = agentpay_client.consume(
        api_key=api_key,
        amount_cents=ALERT_COST_CENTS,
        usage_event_id=usage_id
    )

    if not result.success:
        return f"Error: {result.error_message}"

    # Rest of the existing get_alerts implementation...
    # [Previous implementation remains the same]

# Update get_forecast tool similarly
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # Get API key from context
    api_key = api_key_context.get()
    if not api_key:
        return "Error: API Key missing"

    # Charge for usage
    usage_id = str(uuid.uuid4())
    result = agentpay_client.consume(
        api_key=api_key,
        amount_cents=FORECAST_COST_CENTS,
        usage_event_id=usage_id
    )

    if not result.success:
        return f"Error: {result.error_message}"

    # Rest of the existing get_forecast implementation...
    # [Previous implementation remains the same]

Your server is now fully integrated with AgentPay, handling API key validation and usage charging.

Next Steps

Learn how to test your server with an MCP Client.
Review the working example repo on GitHub.

Get Started

MCP Server Developers

MCP Client Users

SDK Reference (Python)

Prerequisites

Step 1: Project Setup

Step 2: Basic Server Implementation

Step 3: Add Remote Server Capabilities

Step 4: Add AgentPay Integration

i. Initialize AgentPayClient

ii. Extract User API Key

iii. Validate User API Key

iv. Consume Usage

Next Steps

​Prerequisites

​Step 1: Project Setup

​Step 2: Basic Server Implementation

​Step 3: Add Remote Server Capabilities

​Step 4: Add AgentPay Integration

​i. Initialize AgentPayClient

​ii. Extract User API Key

​iii. Validate User API Key

​iv. Consume Usage

​Next Steps

Prerequisites

Step 1: Project Setup

Step 2: Basic Server Implementation

Step 3: Add Remote Server Capabilities

Step 4: Add AgentPay Integration

i. Initialize AgentPayClient

ii. Extract User API Key

iii. Validate User API Key

iv. Consume Usage

Next Steps