98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Clean Proxys MCP server — exposes the proxy store as MCP tools for agents.
|
|
|
|
Wraps the Clean Proxys REST API (same host). Run as streamable-http on :8090/mcp
|
|
so any MCP client (Hermes, Claude, GPTs) can discover and buy proxies natively.
|
|
"""
|
|
import json
|
|
import os
|
|
|
|
import httpx
|
|
from mcp.server import FastMCP
|
|
|
|
API = os.environ.get("CLEAN_PROXYS_API", "http://127.0.0.1:5000")
|
|
|
|
mcp = FastMCP(
|
|
"clean-proxys",
|
|
host="0.0.0.0", # disable DNS-rebinding guard so LAN/remote agents can connect
|
|
instructions=(
|
|
"Clean Proxys — buy clean residential SOCKS5/HTTP proxies programmatically. "
|
|
"No KYC, Bitcoin billing, global geo-targeting. "
|
|
"Flow: list_plans -> buy_proxy -> poll get_order_status until paid -> "
|
|
"connect using the returned socks5/http strings."
|
|
),
|
|
)
|
|
|
|
|
|
def _get(path, params=None):
|
|
r = httpx.get(f"{API}{path}", params=params, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def _post(path, body):
|
|
r = httpx.post(f"{API}{path}", json=body, timeout=25)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
@mcp.tool()
|
|
def get_info() -> str:
|
|
"""Service info, value proposition, proxy ports, and billing model."""
|
|
return json.dumps(_get("/api/v1/info"), indent=2)
|
|
|
|
|
|
@mcp.tool()
|
|
def list_plans() -> str:
|
|
"""List residential GB plans (id, name, gb, price_sats)."""
|
|
return json.dumps(_get("/api/v1/plans"), indent=2)
|
|
|
|
|
|
@mcp.tool()
|
|
def buy_proxy(
|
|
plan_id: int,
|
|
country: str = "",
|
|
city: str = "",
|
|
username: str = "",
|
|
password: str = "",
|
|
) -> str:
|
|
"""One-call purchase: creates an account + Bitcoin invoice and returns working
|
|
proxy credentials plus a checkout URL.
|
|
|
|
Args:
|
|
plan_id: plan id from list_plans.
|
|
country: ISO country code (US, GB, JP, ...) for geo-targeting; empty = any.
|
|
city: optional city.
|
|
username: optional account username (auto-generated if empty).
|
|
password: optional account password (auto-generated if empty).
|
|
|
|
Returns socks5/http connection strings, the proxy password, and an invoice
|
|
checkout_url plus status_url to poll for payment.
|
|
"""
|
|
body = {
|
|
"plan_id": plan_id,
|
|
"country": country,
|
|
"city": city,
|
|
"username": username,
|
|
"password": password,
|
|
}
|
|
return json.dumps(_post("/api/v1/proxy", body), indent=2)
|
|
|
|
|
|
@mcp.tool()
|
|
def get_order_status(order_id: str) -> str:
|
|
"""Poll the payment status of a proxy order (pending or paid)."""
|
|
return json.dumps(_get(f"/api/v1/order/{order_id}"), indent=2)
|
|
|
|
|
|
@mcp.tool()
|
|
def list_dedicated_locations(product: int = 9) -> str:
|
|
"""List dedicated IP locations. product: 9=ISP, 3=datacenter."""
|
|
return json.dumps(_get("/dedicated/locations", params={"product": product}), indent=2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.settings.host = "0.0.0.0"
|
|
mcp.settings.port = int(os.environ.get("MCP_PORT", "8090"))
|
|
mcp.run(transport="streamable-http", mount_path="/mcp")
|