Skip to main content

​
Proxy

MCP Rank provides two types of proxy access:
  1. OAuth Proxy - Call Google APIs (Gmail, Calendar, Drive) on behalf of users
  2. MCP Server Proxy - Call MCP server tools through MCP Rank
Both proxies handle authentication, rate limiting, and error normalization for you.

​
OAuth Proxy (Google APIs)

Make API calls to Google services using stored OAuth tokens. You never see the tokens - just provide the user’s MIT.

​
Prerequisites

Before making proxy requests, ensure:
  1. Client is initialized with user_id
  2. User has connected Google (via JIT auth or manual flow)
from mcp_rank import MCPRankClient

client = MCPRankClient(
    api_key="sk_mcp_rank_...",
    user_id="user_123",
)
await client.initialize()

​
Check Access

Before making requests, verify the user has connected the provider: 
info = await client.get_proxy_info("google")

if info["can_access"]:
    # User has connected Google - make requests
    profile = await client.proxy_get("google", "gmail/v1/users/me/profile")
else:
    # User needs to connect Google first
    session = await client.create_auth_session("google")
    print(f"Please authenticate: {session['auth_url']}")

​
GET Requests

# Gmail profile
profile = await client.proxy_get("google", "gmail/v1/users/me/profile")

# With query parameters
messages = await client.proxy_get(
    "google",
    "gmail/v1/users/me/messages",
    params={"maxResults": "10", "q": "is:unread"}
)

​
POST Requests

# Send an email
import base64
from email.mime.text import MIMEText

message = MIMEText("Hello from MCP Rank!")
message["to"] = "recipient@example.com"
message["subject"] = "Test Email"
raw = base64.urlsafe_b64encode(message.as_bytes()).decode()

result = await client.proxy_post(
    "google",
    "gmail/v1/users/me/messages/send",
    json={"raw": raw}
)

# Create a calendar event
event = {
    "summary": "Team Meeting",
    "start": {"dateTime": "2024-01-15T10:00:00Z"},
    "end": {"dateTime": "2024-01-15T11:00:00Z"}
}
result = await client.proxy_post(
    "google",
    "calendar/v3/calendars/primary/events",
    json=event
)

​
PUT/PATCH/DELETE

# Update an event
await client.proxy_patch(
    "google",
    f"calendar/v3/calendars/primary/events/{event_id}",
    json={"summary": "Updated Meeting Title"}
)

# Delete an event
await client.proxy_delete(
    "google",
    f"calendar/v3/calendars/primary/events/{event_id}"
)

​
Google API Examples

​
Gmail

# List messages
messages = await client.proxy_get(
    "google", "gmail/v1/users/me/messages",
    params={"maxResults": "20"}
)

# Get a specific message
message = await client.proxy_get(
    "google", f"gmail/v1/users/me/messages/{message_id}"
)

# Search emails
results = await client.proxy_get(
    "google", "gmail/v1/users/me/messages",
    params={"q": "from:boss@company.com is:unread"}
)

​
Calendar

# List calendars
calendars = await client.proxy_get(
    "google", "calendar/v3/users/me/calendarList"
)

# Get upcoming events
from datetime import datetime
events = await client.proxy_get(
    "google", "calendar/v3/calendars/primary/events",
    params={
        "timeMin": datetime.utcnow().isoformat() + "Z",
        "maxResults": "10",
        "singleEvents": "true",
        "orderBy": "startTime"
    }
)

​
Drive

# List files
files = await client.proxy_get(
    "google", "drive/v3/files",
    params={"pageSize": "10"}
)

# Search files
results = await client.proxy_get(
    "google", "drive/v3/files",
    params={"q": "name contains 'report'"}
)

​
MCP Server Proxy

Call MCP server tools through MCP Rank. This abstracts away:
  • Transport: SSE responses are buffered to JSON
  • Server URLs: Just use the server ID
  • Authentication: Server-specific auth handled for you

​
Get Server Info

# Check server availability
response = await client._client.get(
    "/v1/mcp/server",
    params={"server_id": "ai.exa/exa"}
)
server = response.json()

print(f"Server: {server['name']}")
print(f"URL: {server['remote_url']}")
print(f"Auth required: {server['auth_required']}")

​
List Server Tools

# Get available tools from an MCP server
response = await client._client.post(
    "/v1/mcp/tools/list",
    json={"server_id": "ai.exa/exa"}
)
tools = response.json()

for tool in tools["tools"]:
    print(f"{tool['name']}: {tool['description']}")

​
Call a Tool

# Call a specific tool
response = await client._client.post(
    "/v1/mcp/tools/call",
    json={
        "server_id": "ai.exa/exa",
        "name": "web_search",
        "arguments": {"query": "latest AI news"}
    }
)
result = response.json()
print(result)

​
Generic JSON-RPC Call

For full control, use the generic call endpoint: 
# Any JSON-RPC method
response = await client._client.post(
    "/v1/mcp/call",
    json={
        "server_id": "ai.exa/exa",
        "method": "tools/call",
        "params": {
            "name": "web_search",
            "arguments": {"query": "latest AI news"}
        },
        "id": 1
    }
)
result = response.json()
# Returns: {"jsonrpc": "2.0", "id": 1, "result": {...}}

​
Available MCP Methods

MethodDescription
tools/listList available tools
tools/callExecute a tool
resources/listList available resources
resources/readRead a resource
prompts/listList available prompts
prompts/getGet a prompt

​
Error Handling

import httpx
from mcp_rank.auth import AuthRequiredError

try:
    result = await client.proxy_get("google", "gmail/v1/users/me/profile")
except AuthRequiredError as e:
    # User needs to authenticate
    print(f"Please authenticate: {e.auth_url}")
except httpx.HTTPStatusError as e:
    if e.response.status_code == 403:
        print("Access denied - token may be expired")
    elif e.response.status_code == 404:
        print("Resource not found")
    elif e.response.status_code == 429:
        print("Rate limited - retry later")
    else:
        print(f"API error: {e.response.text}")

​
Rate Limits

Proxy requests are subject to:
  1. MCP Rank rate limits (based on your API key tier)
  2. Provider rate limits (Google API quotas, MCP server limits)
Handle rate limits gracefully: 
import asyncio

async def with_retry(fn, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await fn()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                wait = 2 ** attempt
                print(f"Rate limited, waiting {wait}s...")
                await asyncio.sleep(wait)
            else:
                raise
    raise Exception("Max retries exceeded")

# Usage
result = await with_retry(
    lambda: client.proxy_get("google", "gmail/v1/users/me/profile")
)