Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.agentpay.me/llms.txt

Use this file to discover all available pages before exploring further.

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

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