# Echo server

> A dependency-free local HTTP server that prints incoming requests to stdout and echoes them back as JSON. Use it as a test target in development and guide walkthroughs.
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_URL is 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.

```python
#!/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

```bash
python3 echo-server.py
```

The server binds to http://0.0.0.0:8080. Set http://localhost:8080 as the Target Endpoint in the gateway dashboard.

### Change the port

```bash
PORT=3000 python3 echo-server.py
```

Set 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.

```bash
RELAY_URL=
 python3 echo-server.py
```

With 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 relay field 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:

```bash
python3 echo-server.py
```

- 
In 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.

Note

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:

```json
{
  "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:

```text
────────────────────────────────────────────────────────────────
  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](/products/affinidi-trust-fabric/agent-gateway/get-started/create-first-surface.md).

- [Restrict surface access with API key authentication](/products/affinidi-trust-fabric/agent-gateway/how-to-guides/access/restrict-surface-access-with-api-key.md).

- [Control MCP tool access with per-tool policies](/products/affinidi-trust-fabric/agent-gateway/how-to-guides/policies/control-mcp-tool-access-with-per-tool-policies.md).

- [Apply an outbound policy to a Transit Point](/products/affinidi-trust-fabric/agent-gateway/how-to-guides/policies/apply-policy-to-outbound-calls.md).