#!/usr/bin/python3 # -*- coding: utf-8 -*- """ HTTP 代理 Tab 提供一个通用的 HTTP 请求代理，既支持在 Gradio 页面上手动调试， 也可以通过 Gradio 的 API 从外部代码调用，例如: # 通过 gradio_client 调用 from gradio_client import Client client = Client("http://127.0.0.1:7860/") status, headers, body = client.predict( url="https://httpbin.org/get", method="GET", headers="{}", params="{}", body="", timeout=30.0, follow_redirects=True, verify_ssl=True, api_name="/http_proxy", ) # 通过 HTTP 调用 gradio 内置 API POST http://127.0.0.1:7860/gradio_api/call/http_proxy body: {"data": ["https://httpbin.org/get", "GET", "{}", "{}", "", 30.0, true, true]} cookies 用法: 没有单独的 cookies 字段, 直接在 headers (json) 里加一个 `Cookie` 头即可, 多个 cookie 用 `;` 分隔, 例如: { "Cookie": "session_id=abc123; user_token=xyz789" } """ import json import logging from typing import Any, Optional import gradio as gr import httpx logger = logging.getLogger("http_proxy") SUPPORTED_METHODS = ["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"] MAX_BODY_PREVIEW = 2 * 1024 * 1024 # 2 MB def _parse_json_field(value: Optional[str], field_name: str) -> Optional[Any]: if value is None: return None text = value.strip() if not text: return None try: return json.loads(text) except json.JSONDecodeError as e: raise ValueError(f"`{field_name}` JSON 解析失败: {e}") from e def http_proxy_request( url: str, method: str = "GET", headers: Optional[str] = None, params: Optional[str] = None, body: Optional[str] = None, timeout: float = 30.0, follow_redirects: bool = True, verify_ssl: bool = True, ): """转发 HTTP 请求并返回 (status_code, response_headers_json, response_body).""" if not url or not url.strip(): return 0, "{}", "url 不能为空" method_upper = (method or "GET").upper() if method_upper not in SUPPORTED_METHODS: return 0, "{}", f"不支持的 method: {method}" try: headers_dict = _parse_json_field(headers, "headers") params_dict = _parse_json_field(params, "params") except ValueError as e: return 0, "{}", str(e) request_kwargs: dict = { "method": method_upper, "url": url.strip(), "headers": headers_dict, "params": params_dict, "timeout": float(timeout) if timeout else 30.0, } if body and body.strip() and method_upper not in ("GET", "HEAD"): body_text = body try: request_kwargs["json"] = json.loads(body_text) except (json.JSONDecodeError, TypeError): request_kwargs["content"] = body_text.encode("utf-8") logger.info(f"proxy request: {method_upper} {url}") try: with httpx.Client( follow_redirects=bool(follow_redirects), verify=bool(verify_ssl), http2=True, ) as client: response = client.request(**request_kwargs) except httpx.HTTPError as e: logger.exception("http error") return 0, "{}", f"http 请求异常: {e!r}" except Exception as e: logger.exception("unexpected error") return 0, "{}", f"未知异常: {e!r}" response_headers = json.dumps( dict(response.headers), ensure_ascii=False, indent=2, ) content_bytes = response.content or b"" if len(content_bytes) > MAX_BODY_PREVIEW: truncated_note = ( f"\n\n... [body 已截断, 原始长度 {len(content_bytes)} bytes, " f"仅展示前 {MAX_BODY_PREVIEW} bytes]" ) content_bytes = content_bytes[:MAX_BODY_PREVIEW] else: truncated_note = "" try: encoding = response.encoding or "utf-8" response_body = content_bytes.decode(encoding, errors="replace") except Exception: response_body = repr(content_bytes) return response.status_code, response_headers, response_body + truncated_note def get_http_proxy_tab(): method_choices = SUPPORTED_METHODS with gr.TabItem("http_proxy"): gr.Markdown( "## HTTP 代理\n" "在页面上手动调试请求，也可以从外部通过 Gradio API 调用，" "把本服务当成统一的 HTTP 代理网关。\n\n" "- gradio_client: `client.predict(..., api_name=\"/http_proxy\")`\n" "- HTTP: `POST /gradio_api/call/http_proxy`，参数顺序见输入框" ) with gr.Row(): with gr.Column(scale=3): http_url = gr.Textbox( label="url", placeholder="https://example.com/api/xxx", lines=1, ) http_method = gr.Dropdown( label="method", choices=method_choices, value=method_choices[0], ) http_headers = gr.Textbox( label="headers (json)", value="{}", lines=4, max_lines=20, ) http_params = gr.Textbox( label="query params (json)", value="{}", lines=4, max_lines=20, ) http_body = gr.Textbox( label="body (raw text 或 json)", value="", lines=6, max_lines=30, ) with gr.Row(): http_timeout = gr.Number(label="timeout", value=30.0) http_follow_redirects = gr.Checkbox( label="follow_redirects", value=True, ) http_verify_ssl = gr.Checkbox( label="verify_ssl", value=True, ) http_send = gr.Button("send", variant="primary") with gr.Column(scale=4): http_status_code = gr.Number(label="status_code", interactive=False) http_response_headers = gr.Textbox( label="response headers", lines=8, max_lines=40, interactive=False, show_copy_button=True, ) http_response_body = gr.Textbox( label="response body", lines=20, max_lines=200, interactive=False, show_copy_button=True, ) inputs = [ http_url, http_method, http_headers, http_params, http_body, http_timeout, http_follow_redirects, http_verify_ssl, ] outputs = [ http_status_code, http_response_headers, http_response_body, ] http_send.click( fn=http_proxy_request, inputs=inputs, outputs=outputs, api_name="http_proxy", ) gr.Examples( examples=[ [ "https://httpbin.org/get", "GET", "{}", "{\"q\": \"hello\"}", "", 30.0, True, True, ], [ "https://httpbin.org/post", "POST", "{\"Content-Type\": \"application/json\"}", "{}", "{\"hello\": \"world\"}", 30.0, True, True, ], [ "https://api.github.com/repos/gradio-app/gradio", "GET", "{\"User-Agent\": \"gradio-http-proxy\"}", "{}", "", 30.0, True, True, ], [ "https://httpbin.org/cookies", "GET", "{\"Cookie\": \"session_id=abc123; user_token=xyz789\"}", "{}", "", 30.0, True, True, ], ], inputs=inputs, ) return locals() if __name__ == "__main__": with gr.Blocks() as block: _ = get_http_proxy_tab() block.queue().launch()