proxywhirl.mcp.server¶

MCP Server implementation for ProxyWhirl.

Production-ready MCP server with unified tool interface, auto-loading from database, proper lifecycle management, FastMCP v2 Context support, and middleware-based auth.

Classes¶

ProxyWhirlMCPServer

ProxyWhirl MCP Server using FastMCP.

Functions¶

`cleanup_rotator`()	Clean up global rotator resources.
`get_auth`()	Get or create the global MCPAuth instance.
`get_proxy_config`()	Wrapper to call shared get_proxy_config implementation.
`get_proxy_health`()	Wrapper to call shared get_proxy_health implementation.
`get_rate_limit_status`(proxy_id[, api_key])	Get rate limiting status for a specific proxy (backward compatibility wrapper).
`get_rotator`()	Get or create the global AsyncProxyWhirl instance with auto-loading.
`list_proxies`([api_key])	Wrapper to call shared list_proxies implementation.
`main`()	CLI entry point for running the MCP server.
`mcp_lifespan`()	Lifespan context manager for MCP server (legacy wrapper).
`proxy_selection_workflow`()	Workflow prompt for guiding proxy selection decisions.
`proxy_status`(proxy_id[, api_key])	Wrapper to call shared proxy_status implementation.
`recommend_proxy`([region, performance, api_key])	Wrapper to call shared recommend_proxy implementation.
`rotate_proxy`([api_key])	Wrapper to call shared rotate_proxy implementation.
`set_auth`(auth)	Set the global MCPAuth instance.
`set_rotator`(rotator)	Set the global AsyncProxyWhirl instance (thread-safe).

Module Contents¶

class proxywhirl.mcp.server.ProxyWhirlMCPServer(proxy_manager=None)[source]¶

ProxyWhirl MCP Server using FastMCP.

This server exposes proxy management functionality to AI assistants via the Model Context Protocol using FastMCP’s decorator-based API.

Example:

# Basic usage
server = ProxyWhirlMCPServer()
server.run()

# With custom rotator
rotator = AsyncProxyWhirl()
server = ProxyWhirlMCPServer(proxy_manager=rotator)
await server.initialize()  # Must call for async setup
server.run()

Initialize ProxyWhirl MCP server.

Note: If providing a proxy_manager, you must call initialize() to set it up.

Parameters:: proxy_manager (Any) – ProxyWhirl proxy manager instance (AsyncProxyWhirl)

async initialize()[source]¶

Initialize the server asynchronously.

Call this method after construction if you provided a custom proxy_manager. This properly sets up the rotator in an async context.

Return type:: None

run(transport='stdio')[source]¶

Run the MCP server.

Parameters:: transport (Literal['stdio', 'http', 'sse', 'streamable-http']) – Transport type (‘stdio’, ‘http’, ‘sse’, or ‘streamable-http’)
Return type:: None

async proxywhirl.mcp.server.cleanup_rotator()[source]¶

Clean up global rotator resources.

Closes HTTP clients and releases all resources held by the rotator. Safe to call multiple times.

Return type:: None

proxywhirl.mcp.server.get_auth()[source]¶

Get or create the global MCPAuth instance.

Returns:: MCPAuth instance
Return type:: proxywhirl.mcp.auth.MCPAuth

async proxywhirl.mcp.server.get_proxy_config()[source]¶

Wrapper to call shared get_proxy_config implementation.

Return type:: str

async proxywhirl.mcp.server.get_proxy_health()[source]¶

Wrapper to call shared get_proxy_health implementation.

Return type:: str

async proxywhirl.mcp.server.get_rate_limit_status(proxy_id, api_key=None)[source]¶

Get rate limiting status for a specific proxy (backward compatibility wrapper).

Note: Rate limiting info is now included in the health action response. This wrapper is kept for backward compatibility.

Parameters:

proxy_id (str) – UUID of the proxy to check
api_key (str | None) – Optional API key for authentication

Returns:

Rate limit information.

Return type:

dict[str, Any]

async proxywhirl.mcp.server.get_rotator()[source]¶

Get or create the global AsyncProxyWhirl instance with auto-loading.

On first initialization: 1. If database exists (from PROXYWHIRL_MCP_DB env or proxywhirl.db), loads proxies 2. If pool is still empty, auto-fetches proxies from public sources

Returns:: AsyncProxyWhirl instance
Return type:: proxywhirl.rotator.AsyncProxyWhirl

async proxywhirl.mcp.server.list_proxies(api_key=None)[source]¶

Wrapper to call shared list_proxies implementation.

Parameters:: api_key (str | None) – Optional API key for authentication
Returns:: Proxy list and pool statistics.
Return type:: dict[str, Any]

proxywhirl.mcp.server.main()[source]¶

CLI entry point for running the MCP server.

This enables uvx proxywhirl[mcp] or proxywhirl-mcp to start the server.

Environment variables:: PROXYWHIRL_MCP_API_KEY: API key for authentication PROXYWHIRL_MCP_DB: Path to proxy database file PROXYWHIRL_MCP_LOG_LEVEL: Log level (debug, info, warning, error)

Return type:: None

async proxywhirl.mcp.server.mcp_lifespan()[source]¶

Lifespan context manager for MCP server (legacy wrapper).

Note: FastMCP v2 uses the lifespan passed to FastMCP() constructor. This is kept for backward compatibility with direct usage.

Yields:: None

async proxywhirl.mcp.server.proxy_selection_workflow()[source]¶

Workflow prompt for guiding proxy selection decisions.

Return type:: str

async proxywhirl.mcp.server.proxy_status(proxy_id, api_key=None)[source]¶

Wrapper to call shared proxy_status implementation.

Parameters:

proxy_id (str) – UUID of the proxy to check
api_key (str | None) – Optional API key for authentication

Returns:

Proxy status and metrics.

Return type:

dict[str, Any]

async proxywhirl.mcp.server.recommend_proxy(region=None, performance='medium', api_key=None)[source]¶

Wrapper to call shared recommend_proxy implementation.

Parameters:

region (str | None) – Optional region filter (country code)
performance (str | None) – Performance tier (high, medium, low)
api_key (str | None) – Optional API key for authentication

Returns:

Recommendation and alternatives.

Return type:

dict[str, Any]

async proxywhirl.mcp.server.rotate_proxy(api_key=None)[source]¶

Wrapper to call shared rotate_proxy implementation.

Parameters:: api_key (str | None) – Optional API key for authentication
Returns:: Selected proxy information.
Return type:: dict[str, Any]

proxywhirl.mcp.server.set_auth(auth)[source]¶

Set the global MCPAuth instance.

Parameters:: auth (proxywhirl.mcp.auth.MCPAuth | None) – MCPAuth instance to use, or None to disable authentication
Return type:: None

async proxywhirl.mcp.server.set_rotator(rotator)[source]¶

Set the global AsyncProxyWhirl instance (thread-safe).

Parameters:: rotator (proxywhirl.rotator.AsyncProxyWhirl) – AsyncProxyWhirl instance to use
Return type:: None