| |
| """FastAPI mock server for debugging proxy/client behavior. |
| |
| This is a development helper (not used by the main KiroProxy app). |
| |
| Typical usage: |
| python scripts/proxy_server.py --port 8000 |
| |
| Then point your client/proxy to `http://127.0.0.1:8000` and inspect requests: |
| GET http://127.0.0.1:8000/logs |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import logging |
| from datetime import datetime |
| from typing import Any, Dict, List |
|
|
| import uvicorn |
| from fastapi import FastAPI, Request |
| from fastapi.responses import JSONResponse |
|
|
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") |
| logger = logging.getLogger("kiro_proxy.mock_server") |
|
|
| app = FastAPI(title="KiroProxy Mock Server") |
| request_log: List[Dict[str, Any]] = [] |
|
|
|
|
| @app.middleware("http") |
| async def log_requests(request: Request, call_next): |
| body = await request.body() |
|
|
| request_log.append( |
| { |
| "timestamp": datetime.now().isoformat(), |
| "method": request.method, |
| "url": str(request.url), |
| "path": request.url.path, |
| "headers": dict(request.headers), |
| "body": body.decode("utf-8", errors="ignore")[:2000] if body else None, |
| } |
| ) |
|
|
| logger.info("%s %s", request.method, request.url.path) |
| return await call_next(request) |
|
|
|
|
| @app.get("/") |
| async def root(): |
| return {"status": "ok", "message": "Mock server running", "requests_logged": len(request_log)} |
|
|
|
|
| @app.get("/logs") |
| async def get_logs(limit: int = 50): |
| return {"total": len(request_log), "requests": request_log[-limit:]} |
|
|
|
|
| @app.get("/clear") |
| async def clear_logs(): |
| request_log.clear() |
| return {"message": "Logs cleared"} |
|
|
|
|
| @app.api_route("/auth/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"]) |
| async def mock_auth(_: Request, path: str): |
| logger.info("Auth request: %s", path) |
| return JSONResponse({"success": True, "token": "mock-token-for-testing", "expires_in": 3600}) |
|
|
|
|
| @app.post("/v1/chat/completions") |
| async def mock_chat_completions(request: Request): |
| body = await request.json() |
| logger.info("Chat body preview: %s", json.dumps(body, ensure_ascii=False)[:500]) |
|
|
| return JSONResponse( |
| { |
| "id": "chatcmpl-test", |
| "object": "chat.completion", |
| "created": int(datetime.now().timestamp()), |
| "model": body.get("model", "mock-model"), |
| "choices": [ |
| { |
| "index": 0, |
| "message": {"role": "assistant", "content": "Mock response from scripts/proxy_server.py"}, |
| "finish_reason": "stop", |
| } |
| ], |
| "usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, |
| } |
| ) |
|
|
|
|
| @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"]) |
| async def catch_all(request: Request, path: str): |
| logger.info("Caught: %s /%s", request.method, path) |
| return JSONResponse( |
| { |
| "proxy_status": "intercepted", |
| "method": request.method, |
| "path": f"/{path}", |
| "message": "Request intercepted by mock server", |
| "headers_received": dict(request.headers), |
| } |
| ) |
|
|
|
|
| def _parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description="KiroProxy mock server (dev helper)") |
| parser.add_argument("--host", default="0.0.0.0", help="Bind host") |
| parser.add_argument("--port", type=int, default=8000, help="Bind port") |
| return parser.parse_args() |
|
|
|
|
| if __name__ == "__main__": |
| args = _parse_args() |
| uvicorn.run(app, host=args.host, port=args.port) |
|
|
