Echo server
A single-file Python 3 HTTP server that accepts any request on any path, prints the full incoming request to stdout, and returns a JSON echo response. No packages beyond the Python 3 standard library are required.
The echo server serves as a local test target for the gateway. Because it runs on your machine and prints to your terminal, you can see exactly what the gateway forwarded: injected headers, modified bodies, and identity context, without interpreting an intermediary’s envelope format.
Two modes are available:
- Passive echo: accepts any request, prints it to stdout, and returns the echo JSON. No outbound traffic is generated. Use this as the target endpoint for any inbound surface.
- Active relay: when
RELAY_URLis set, the server forwards the incoming request to that URL and returns the relay response transparently to the caller, including status code, content type, and body. Both legs are still printed to stdout. Use this for end-to-end surface testing where the caller should receive the Transit Point’s actual response.
Script
Copy the script below, save it as echo-server.py, and run it with python3 echo-server.py. No pip installs required. Python 3.7+ standard library only.
#!/usr/bin/env python3
import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
PORT = int(os.environ.get("PORT", "8080"))
RELAY_URL = os.environ.get("RELAY_URL", "").strip()
class EchoHandler(BaseHTTPRequestHandler):
server_version = "EchoServer/1.0"
protocol_version = "HTTP/1.1"
def log_message(self, fmt, *args):
pass
def _read_body(self):
length = int(self.headers.get("Content-Length", 0))
if length > 0:
return self.rfile.read(length).decode("utf-8", errors="replace")
return ""
def _headers_as_dict(self):
return {k: v for k, v in self.headers.items()}
def _print_incoming(self, method, body):
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
bar = "─" * 64
print(f"\n{bar}")
print(f" {ts} {method} {self.path}")
print(bar)
for k, v in self.headers.items():
print(f" {k}: {v}")
if body:
print(bar)
try:
print(json.dumps(json.loads(body), indent=2))
except json.JSONDecodeError:
print(body)
print(bar, flush=True)
def _relay_request(self, method, body, request_headers):
data = body.encode("utf-8") if body else None
req = urllib.request.Request(RELAY_URL, data=data, method=method)
skip = {"host", "content-length", "transfer-encoding", "connection"}
for k, v in request_headers.items():
if k.lower() not in skip:
req.add_header(k, v)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return (
resp.status,
resp.read().decode("utf-8", errors="replace"),
resp.headers.get("Content-Type", "application/json"),
)
except urllib.error.HTTPError as exc:
return exc.code, exc.read().decode("utf-8", errors="replace"), "application/json"
except Exception as exc:
return 0, f"relay error: {exc}", "text/plain"
def _handle(self):
method = self.command
body = self._read_body()
self._print_incoming(method, body)
echo = {
"echo": {
"method": method,
"path": self.path,
"headers": self._headers_as_dict(),
"body": body,
}
}
if RELAY_URL:
status, relay_body, content_type = self._relay_request(method, body, self.headers)
print(f" → relay {RELAY_URL} → HTTP {status}", flush=True)
payload = relay_body.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
payload = json.dumps(echo, indent=2).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = do_OPTIONS = _handle
def main():
server = HTTPServer(("0.0.0.0", PORT), EchoHandler)
relay_line = (
f" Relay URL : {RELAY_URL}"
if RELAY_URL
else " Relay URL : (not set — passive echo mode)"
)
print(
f"\nEcho server running.\n"
f" Listening on : http://0.0.0.0:{PORT}\n"
f"{relay_line}\n\n"
f"Incoming requests are printed here and echoed back as JSON.\n"
f"Press Ctrl+C to stop.\n",
flush=True,
)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopped.", flush=True)
sys.exit(0)
if __name__ == "__main__":
main()Environment variables
| Variable | Default | Description |
|---|---|---|
PORT | 8080 | TCP port the server listens on. |
RELAY_URL | (not set) | When set, every inbound request is forwarded to this URL after echoing. The relay response appears in the relay field of the echo body. Leave unset for passive echo mode. |
Usage
Start in passive echo mode
python3 echo-server.pyThe server binds to http://0.0.0.0:8080. Set http://localhost:8080 as the Target Endpoint in the gateway dashboard.
Change the port
PORT=3000 python3 echo-server.pySet http://localhost:3000 as the Target Endpoint in the dashboard.
Start in relay mode for Transit Point testing
Replace <TRANSIT-POINT-URL> with the Listener URL shown in the Transit Point config panel.
RELAY_URL=<TRANSIT-POINT-URL> python3 echo-server.pyWith this configuration:
- The gateway forwards inbound requests to
http://localhost:8080(the echo server). - The echo server prints the request to stdout, then forwards it to the Transit Point URL.
- The Transit Point response from the gateway appears in the
relayfield of the echo body.
This lets you observe both legs, inbound forwarding and outbound relay, from a single terminal.
Expose to a remote gateway
If your gateway runs on a remote host, it cannot reach localhost on your machine directly. Use a tunnelling tool to expose the echo server at a public HTTPS URL.
Start the echo server:
python3 echo-server.pyIn a second terminal, start a tunnel on port 8080 using your chosen tunnelling tool.
Copy the public
https://forwarding URL that the tool provides.Set that URL as the Target Endpoint in the gateway dashboard instead of
http://localhost:8080.
A tunnelling tool is only needed when the gateway cannot reach your local machine directly. For a gateway running on the same machine as the echo server, http://localhost:8080 is sufficient.
Response format
Passive echo mode: every request returns 200 OK with a Content-Type: application/json body:
{
"echo": {
"method": "POST",
"path": "/agents/my-surface",
"headers": {
"content-type": "application/json",
"x-api-key": "my-test-key",
"x-forwarded-for": "203.0.113.1"
},
"body": "{\"jsonrpc\":\"2.0\",\"method\":\"message/send\",\"id\":\"test-001\",\"params\":{...}}"
}
}Active relay mode: the gateway returns the Transit Point’s response transparently. The HTTP status code, Content-Type, and body are exactly what the Transit Point returned. Both the inbound request and the relay result are printed to the terminal for inspection.
Terminal output
Each request is printed to stdout as it arrives:
────────────────────────────────────────────────────────────────
2026-07-08T09:14:02Z POST /agents/my-surface
────────────────────────────────────────────────────────────────
host: localhost:8080
content-type: application/json
x-api-key: my-test-key
────────────────────────────────────────────────────────────────
{
"jsonrpc": "2.0",
"method": "message/send",
"id": "test-001",
"params": { ... }
}
────────────────────────────────────────────────────────────────Related
- Create your first surface.
- Restrict surface access with API key authentication.
- Control MCP tool access with per-tool policies.
- Apply an outbound policy to a Transit Point.
Glad to hear it! Please tell us how we can improve more.
Sorry to hear that. Please tell us how we can improve.
Thank you for sharing your feedback so we can improve your experience.