Use FastAPI to Wrap a Highly Available Proxy to Obtain API Services

As web data grows in scale, more and more Python developers, web crawlers, and backend service developers need to use HTTP proxy services in their business operations to overcome limitations, simulate real user access, and achieve high-sequence crawling goals. This article will take a practical approach and explain in detail how to use FastAPI to encapsulate a clear and maintainable proxy retrieval interface service, allowing your web crawlers or microservices to “transparently use the proxy” without frequently rewriting the calling logic.

This article is suitable for:

• A backend engineer proficient in Python development or Flask/Django.
• Developers who need to build API interfaces to provide proxy services to external parties
• The goal is to build a web crawler framework that provides a unified management system for proxy IP calls.

In this example, we will build a scalable service that has:

• The standard REST API interface is used to obtain the proxy.
• Request management based on middleware/dependency injection
• Decoupling the proxy request logic from the business layer improves maintainability.
• Compatible with real residential IP proxy solutions (such as kookeey static residential proxy & dynamic residential proxy )

1. Core Functionality: How does FastAPI encapsulate the proxy to obtain the interface?

This section guides you through building a working proxy interface service from scratch, enabling any business application to automatically initiate HTTP requests through the proxy simply by calling an API.

Project initialization

mkdir fastapi-proxy-service

cd fastapi-proxy-service

python3 -m venv venv

source venv/bin/activate

pip install fastapi uvicorn httpx pydantic

Define the data model schema.py

from pydantic import BaseModel

class ProxyRequest(BaseModel):

url: str

method: str = "GET"

timeout: int = 10

class ProxyResponse(BaseModel):

status_code: int

content: str

proxy_used: str

error: str = ""

The proxy logic is encapsulated in proxy_manager.py

import httpx, random

PROXY_POOL = [

"http://127.0.0.1:8001",

http://127.0.0.1:8002

]

def choose_proxy():

return random.choice(PROXY_POOL)

async def fetch_via_proxy(url, method="GET", timeout=10):

proxy = choose_proxy()

proxies = {"http://": proxy, "https://": proxy}

async with httpx.AsyncClient(proxies=proxies, timeout=timeout) as client:

resp = await client.request(method, url)

return {

"status_code": resp.status_code,

"content": resp.text,

"proxy_used": proxy

}

service layer service.py

from fastapi import HTTPException

from app.schema import ProxyRequest, ProxyResponse

from app.proxy_manager import fetch_via_proxy

async def handle_proxy_request(req: ProxyRequest) -> ProxyResponse:

try:

result = await fetch_via_proxy(req.url, req.method, req.timeout)

return ProxyResponse(**result)

except Exception as e:

raise HTTPException(status_code=500, detail=str(e))

The routing entry point is main.py

import FastAPI from fastapi

from app.schema import ProxyRequest, ProxyResponse

from app.service import handle_proxy_request

app = FastAPI()

@app.post("/proxy", response_model=ProxyResponse)

async def proxy(req: ProxyRequest):

return await handle_proxy_request(req)

At this point, a basic, working proxy API service has been completed.

2. Upgrading Agency Capabilities: Connecting with High-Quality Residential Agencies and Real-World Business Scenarios

To truly implement proxy services in business operations, relying solely on a local proxy pool is insufficient. This section will cover the following aspects: why introduce professional agents, how to choose different agent types, and how kookeey can be seamlessly integrated with the system.

Why introduce professional agents?

Common issues encountered by regular agents:

• The IP address is unstable and easily blocked.
• Low anonymity, unable to simulate real users
• Extremely prone to crashing under high concurrency

In comparison, residential proxy services most closely resemble real user traffic and are the standard solution adopted by most companies. kookeey’s dynamic residential proxy supports 41+ countries and regions worldwide, allows for precise IP control, and adapts to different business needs. Click below for a free trial:

Dynamic Residential Agent: Automatic Rotation Access Example

Example code (without exposing the real API):

async def fetch_kookeey_ip():

resp = await httpx.get(

"https://api.kookeey.com/residential/dynamic/get-ip",

params={"country": "US", "count": 1}

)

return resp.json()["ip_list"]

Join the proxy pool:

async def refill_pool():

new_ips = await fetch_kookeey_ip()

PROXY_POOL.extend(f"http://{ip}" for ip in new_ips)

Static residential agency: bundled business access

STATIC_PROXY = "http://username:password@gateway.kookeey.com:12345"

async def fetch_static(url):

async with httpx.AsyncClient(proxies=STATIC_PROXY) as client:

return (await client.get(url)).text

Best practice: Dependency Injection Proxy

async def get_proxy():

return choose_proxy()

@app.post("/proxy")

async def proxy(req: ProxyRequest, proxy = Depends(get_proxy)):

return await fetch_via_proxy(req.url, method=req.method, timeout=req.timeout)

3. Engineering Optimization: Ensuring Services Sufficient to Support the Production Environment

Performance optimization: Connection pool and lifecycle

app = FastAPI()

@app.on_event("startup")

async def startup():

app.state.client = httpx.AsyncClient(

limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)

)

@app.on_event("shutdown")

async def shutdown():

await app.state.client.aclose()

High availability of proxy pools: health checks and self-healing

async def health_check():

while True:

for proxy in list(PROXY_POOL):

try:

await app.state.client.get("https://httpbin.org/ip", proxies={"http": proxy})

except:

PROXY_POOL.remove(proxy)

await asyncio.sleep(30)

Architecture Upgrade: From Tools to Proxy Gateway

Once your FastAPI service is integrated with enterprise applications, it will evolve into a “proxy gateway” with the following capabilities:

• Unified agency strategy.
• Multi-service reuse capability.
• Observability and access tracking.
• Scalable scheduling capabilities (different services use different proxy pools).

4. Production Implementation: Logs, Tracking IDs, and Authentication

These capabilities are essential for deployment in a production environment.

Unified Log

logger.info({

"event": "proxy_request",

"url": req.url,

"proxy": result.proxy_used,

"status": result.status_code

})

Injection Request ID

@app.middleware("http")

async def add_request_id(request, call_next):

request.state.request_id = str(uuid.uuid4())

response = await call_next(request)

response.headers["X-Request-ID"] = request.state.request_id

return response

API Authentication

def verify_token(token: str):

if token != "YOUR_TOKEN":

raise HTTPException(status_code=401, detail="Invalid token")

Conclusion: From Simple Proxy Interfaces to Enterprise-Level Proxy Platforms

Through this refactoring, your FastAPI proxy service has evolved from a “workable little tool” into a service with:

• High-availability proxy pool
• Dynamic residential agent automatic rotation
• Enterprise-level performance optimization
• Production-level logging and authentication system

Especially after integrating kookeey ‘s static and dynamic residential proxy, the service will see a qualitative improvement in stability, success rate, and cross-regional access capabilities, fully supporting enterprise-level business scenarios.