OCPI server

OCPI is symmetric. You are a server to your roaming partners and a client to them, often in the same request. EVPanda records both sides. Shared setup (install, API key, config, shutdown, troubleshooting) is in Getting started.

Model

The direction follows from the method you call, so there is no argument to get backwards.

Method	Direction	You are the	Typical case
`captureInboundMessage`	`IN`	server	A partner pushes a CDR to your endpoint
`captureOutboundMessage`	`OUT`	client	You pull a partner’s locations

Identity is always the partner on the other side, in both directions, never your own platform.

Adapters

The adapters assemble the exchange for you: collect headers and bodies, resolve the partner, call the right method. A request with no resolvable identity is served exactly as it would have been, just not captured.

Go
Node
Python

Adapter	Direction	For
`ocpi.Middleware(client)`	inbound	net/http, chi, gorilla/mux, `echo.WrapMiddleware`, `gin.WrapH`
`ocpi.RoundTripper(client, base)`	outbound	any `http.Client`. A nil base means `http.DefaultTransport`

import "github.com/evpanda-labs/evpanda-go/ocpi"

srv := &http.Server{Handler: auth(ocpi.Middleware(panda)(mux))}
client := &http.Client{Transport: ocpi.RoundTripper(panda, nil)}

Adapter	Direction	For
`ocpi.express(client)`	inbound	express, connect, or a bare `node:http` server
`ocpi.fetch(client, fetch)`	outbound	global `fetch`, or any implementation you pass
`ocpi.axios(client, instance)`	outbound	an axios instance

import { ocpi, startOCPI } from "@evpanda/sdk";

const panda = startOCPI();

app.use(ocpi.express(panda));                       // inbound
const fetch = ocpi.fetch(panda, globalThis.fetch);  // outbound
const http = ocpi.axios(panda, axios.create());     // outbound

Adapter	Direction	For
`ocpi.wsgi.WSGIMiddleware`	inbound	Flask, Django, Pyramid, any WSGI app
`ocpi.asgi.ASGIMiddleware`	inbound	FastAPI, Starlette, Litestar, any ASGI app
`ocpi.httpx_transport.HTTPXTransport`	outbound	`httpx`, with `AsyncHTTPXTransport` for async
`ocpi.requests_adapter.RequestsAdapter`	outbound	`requests`

from evpanda.ocpi.asgi import ASGIMiddleware
from evpanda.ocpi.wsgi import WSGIMiddleware

app = ASGIMiddleware(app, panda)                    # FastAPI, Starlette
app.wsgi_app = WSGIMiddleware(app.wsgi_app, panda)  # Flask

import httpx
from evpanda.ocpi.httpx_transport import HTTPXTransport

client = httpx.Client(transport=HTTPXTransport(panda))

from evpanda.ocpi.requests_adapter import instrument_session

session = instrument_session(requests.Session(), panda)

Telling the adapters who the partner is

Your OCPI server already knows: you looked the partner up by their Token A or Token B to decide whether to serve the request. Stamp the identity there. Each SDK reads three carriers in order, and falls back to the X-EVPanda-* headers last.

Go
Node
Python

Stamp the request context in the middleware that authenticates the partner.

func auth(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        partner, ok := lookupPartner(r.Header.Get("Authorization"))
        if !ok {
            http.Error(w, "unknown partner", http.StatusUnauthorized)
            return
        }
        ctx := ocpi.ContextWithIdentity(r.Context(), evpanda.Platform{
            ID:   partner.ID,
            Name: partner.Name,
        })
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

Outbound, stamp the context on the request you are about to send:

ctx = ocpi.ContextWithIdentity(ctx, evpanda.Platform{
    ID: partner.ID, Name: partner.Name,
})
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, partner.URL, body)
req.Header.Set("Authorization", "Token "+partner.TokenB)

resp, err := client.Do(req)
if err != nil {
    return err
}
defer resp.Body.Close() // capture completes when the body is closed

Mount order matters in Go. The middleware resolves identity before it calls the next handler, so your auth layer has to run outside it: auth(ocpi.Middleware(panda)(mux)). Wrapped the other way around, no request has an identity yet and nothing is captured.

Stamp the request object with setIdentity.

import { ocpi } from "@evpanda/sdk";

app.use((req, res, next) => {
  const partner = lookupPartner(req.headers.authorization);
  if (!partner) return res.status(401).json({ status_code: 2001 });
  ocpi.setIdentity(req, { id: partner.id, name: partner.name });
  next();
});

The request is read when the response finishes, so mount order does not matter. Your auth layer can sit inside or outside the capture middleware.Outbound, scope the call with useIdentity:

const response = await ocpi.useIdentity(
  { id: partner.id, name: partner.name },
  () =>
    fetch(`${partner.url}/ocpi/2.2/sessions`, {
      method: "POST",
      headers: { authorization: `Token ${partner.tokenB}` },
      body: JSON.stringify(payload),
    }),
);

It is an AsyncLocalStorage underneath, so it follows the async call chain rather than leaking to whatever else the event loop is running.

Stamp the request mapping with set_identity: the ASGI scope, or the WSGI environ.

import evpanda
from evpanda.ocpi import set_identity

async def authenticate(request, call_next):
    partner = lookup_partner(request.headers.get("authorization"))
    if partner is None:
        return JSONResponse({"status_code": 2001}, status_code=401)
    set_identity(request.scope, evpanda.Platform(
        id=partner.id, name=partner.name,
    ))
    return await call_next(request)

The mapping is read when the response finishes, so it does not matter where you stamp it. A service that authenticates inside the route handler can stamp it there.Outbound, use the context manager:

from evpanda.ocpi import use_identity

with use_identity(evpanda.Platform(id=partner.id, name=partner.name)):
    response = client.post(
        f"{partner.url}/ocpi/2.2/sessions",
        json=payload,
        headers={"Authorization": f"Token {partner.token_b}"},
    )

It is a ContextVar, so it is task local in async code and thread local in sync code. httpx also takes it per request:

client.get(url, extensions={"evpanda.identity": identity})

The X-EVPanda headers

If nothing else carries the identity, every adapter reads these, matched case insensitively.

Header	Field
`X-EVPanda-Platform-Id`	`id`, required
`X-EVPanda-Platform-Name`	`name`, required
`X-EVPanda-Tenant-Id`	`tenantId`, optional pair
`X-EVPanda-Tenant-Name`	`tenantName`, optional pair

Partners never send them and never receive them: the outbound adapters strip all four before dispatch.

Custom resolvers

If identity lives somewhere else entirely, a client certificate or a path prefix, pass your own resolver. Returning nothing means the exchange is not captured, and the request is never blocked or altered.

byPath := func(r *http.Request) (evpanda.Platform, bool) {
    name, ok := strings.CutPrefix(r.URL.Path, "/partners/")
    if !ok {
        return evpanda.Platform{}, false // not captured
    }
    return evpanda.Platform{ID: name, Name: name}, true
}

ocpi.Middleware(panda, ocpi.WithResolver(byPath))
ocpi.RoundTripper(panda, nil, ocpi.WithResolver(byPath))

const byPath: ocpi.Resolver = (info) => {
  if (!info.url.startsWith("/partners/")) return undefined; // not captured
  const name = info.url.split("/")[2];
  return { id: name, name };
};

app.use(ocpi.express(panda, { resolver: byPath }));

from evpanda.ocpi import RequestInfo

def by_path(info: RequestInfo) -> evpanda.Platform | None:
    if not info.url.startswith("/partners/"):
        return None                      # not captured
    name = info.url.removeprefix("/partners/").split("/")[0]
    return evpanda.Platform(id=name, name=name)

ASGIMiddleware(app, panda, resolver=by_path)

Complete example

Go
Node
Python

server.go

package main

import (
    "log"
    "net/http"
    "time"

    evpanda "github.com/evpanda-labs/evpanda-go"
    "github.com/evpanda-labs/evpanda-go/ocpi"
)

func main() {
    panda, err := evpanda.StartOCPI(evpanda.OCPIConfig{})
    if err != nil {
        log.Printf("evpanda: %v (running inert)", err)
    }
    defer func() { _ = panda.Close() }()

    mux := http.NewServeMux()
    mux.Handle("POST /ocpi/2.2/cdrs", cdrHandler)
    mux.Handle("PUT /ocpi/2.2/sessions/{id}", sessionHandler)

    // auth outside, capture inside.
    srv := &http.Server{
        Addr:    ":8080",
        Handler: auth(ocpi.Middleware(panda)(mux)),
    }

    // Outbound calls to partners.
    partners := &http.Client{
        Transport: ocpi.RoundTripper(panda, nil),
        Timeout:   30 * time.Second,
    }
    _ = partners

    log.Fatal(srv.ListenAndServe())
}

func auth(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        partner, ok := lookupPartner(r.Header.Get("Authorization"))
        if !ok {
            http.Error(w, "unknown partner", http.StatusUnauthorized)
            return
        }
        ctx := ocpi.ContextWithIdentity(r.Context(), evpanda.Platform{
            ID:   partner.ID,
            Name: partner.Name,
        })
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

Other routers, since the middleware is the stdlib shape:

r.Use(auth); r.Use(ocpi.Middleware(panda))    // chi
r.Use(auth, ocpi.Middleware(panda))           // gorilla/mux
e.Use(echo.WrapMiddleware(auth))              // echo
e.Use(echo.WrapMiddleware(ocpi.Middleware(panda)))

// gin: an engine is an http.Handler, so wrap it with auth outside.
srv := &http.Server{Handler: auth(ocpi.Middleware(panda)(engine))}

server.ts

import express from "express";
import axiosLib from "axios";
import { ocpi, startOCPI } from "@evpanda/sdk";

const panda = startOCPI();
if (panda.error) log.warn(`${panda.error.message} (running inert)`);

const app = express();

app.use(express.json());        // a body parser must come first
app.use(ocpi.express(panda));   // mount order is otherwise free

app.use((req, res, next) => {
  const partner = lookupPartner(req.headers.authorization);
  if (!partner) return res.status(401).json({ status_code: 2001 });
  ocpi.setIdentity(req, { id: partner.id, name: partner.name });
  next();
});

app.post("/ocpi/2.2/cdrs", handleCdr);

// Outbound. Instrument whichever client you actually use.
export const fetch = ocpi.fetch(panda, globalThis.fetch);
export const partnerApi = ocpi.axios(
  panda,
  axiosLib.create({ baseURL: "https://partner.example" }),
);

process.on("SIGTERM", async () => {
  await panda.close();
  process.exit(0);
});

Because ocpi.fetch returns a plain fetch, any client that accepts one is instrumented too:

const api = ky.create({ fetch });        // ky
const $api = ofetch.create({ fetch });   // ofetch

app.py

import evpanda
from evpanda.ocpi import set_identity
from evpanda.ocpi.asgi import ASGIMiddleware
from fastapi import FastAPI
from starlette.responses import JSONResponse

panda = evpanda.start_ocpi()
if panda.error:
    log.warning("%s (running inert)", panda.error)

app = FastAPI()


@app.middleware("http")
async def authenticate(request, call_next):
    partner = lookup_partner(request.headers.get("authorization"))
    if partner is None:
        return JSONResponse({"status_code": 2001}, status_code=401)
    set_identity(request.scope, evpanda.Platform(
        id=partner.id, name=partner.name,
    ))
    return await call_next(request)


@app.post("/ocpi/2.2/cdrs")
async def cdrs(payload: dict) -> dict:
    ...


# Wrap the finished app. Mount order does not matter.
app = ASGIMiddleware(app, panda)

client.py

import httpx
from evpanda.ocpi import use_identity
from evpanda.ocpi.httpx_transport import AsyncHTTPXTransport

client = httpx.AsyncClient(transport=AsyncHTTPXTransport(panda))


async def push_session(partner, payload):
    with use_identity(evpanda.Platform(id=partner.id, name=partner.name)):
        return await client.post(
            f"{partner.url}/ocpi/2.2/sessions",
            json=payload,
            headers={"Authorization": f"Token {partner.token_b}"},
        )

If your httpx client sets verify, cert, limits, proxy, or http2, pass a base transport carrying them. httpx applies those arguments only when it builds its own transport, so supplying transport= silently drops them, including the TLS ones.

base = httpx.HTTPTransport(verify=ctx, limits=httpx.Limits(max_connections=50))
client = httpx.Client(transport=HTTPXTransport(panda, base))

Client level settings that are not transport arguments, such as timeout, follow_redirects, headers, and auth, are unaffected.

Per-SDK notes

Go: auth must wrap capture

The middleware resolves identity up front, before the handler runs, so the auth layer that stamps the context has to be outside it. Node and Python read their carrier when the response finishes, which is why mount order is free there.

Node: mount a body parser first

The express adapter reads the request body from req.body rather than teeing the raw stream, since a data listener would flip the stream to flowing mode and could starve your own parser. Without a parser, exchanges are captured with no request body.express.raw() gives the exact bytes off the wire. express.json() gives a re-serialized form, which is usually fine and occasionally differs in key order or whitespace.

Node: axios and fetch are separate worlds

In Node, axios goes through node:http and never fetch. Wrapping one captures nothing from the other. Instrument whichever client your OCPI calls actually use, or both.

Python: the WSGI adapter reads the body up front

WSGI gives no way to observe a body the handler never reads, so the WSGI middleware takes it at the start of the request. That is a copy per in-flight request, bounded by max_capture_bytes. The ASGI adapter does not pay it.

Close your response bodies

Outbound capture completes when the response body is read to the end or closed. A leaked body is a message that never ships. A call that fails at the transport layer captures nothing, because there was no exchange.

Capturing without an adapter

When no adapter fits, build the exchange yourself.

data := evpanda.HTTPExchange{
    Method:          "POST",
    URL:             "/ocpi/2.2/cdrs",
    StatusCode:      201,
    RequestHeaders:  map[string]string{"content-type": "application/json"},
    ResponseHeaders: map[string]string{"content-type": "application/json"},
}
data.SetRequestBody(reqBytes)   // copies, so reuse your buffer right after
data.SetResponseBody(respBytes)

panda.CaptureInboundMessage(evpanda.OCPIMessageInput{
    Identity: evpanda.Platform{ID: "acme", Name: "Acme Mobility"},
    Data:     data,
})

panda.captureInboundMessage({
  identity: { id: "acme", name: "Acme Mobility" },
  data: {
    method: "POST",
    url: "/ocpi/2.2/cdrs",
    statusCode: 201,
    requestHeaders: { "content-type": "application/json" },
    responseHeaders: { "content-type": "application/json" },
    requestBody: reqBytes,     // Uint8Array or string
    responseBody: respBytes,
  },
});

panda.capture_inbound_message(
    identity=evpanda.Platform(id="acme", name="Acme Mobility"),
    data=evpanda.HTTPExchange(
        method="POST",
        url="/ocpi/2.2/cdrs",
        status_code=201,
        request_headers={"content-type": "application/json"},
        response_headers={"content-type": "application/json"},
        request_body=request_bytes,   # bytes or str
        response_body=response_bytes,
    ),
)

captureOutboundMessage is the same call for the other direction. Both are non-blocking and never raise. In Go, use SetRequestBody and SetResponseBody rather than assigning the fields: they copy, so you can reuse a pooled buffer as soon as the call returns.

Always set a status code. The SDKs let you omit it, but the ingestion API requires a value between 100 and 599 and rejects the message without one, after a successful delivery. For an exchange that never produced a response, skip the capture instead.

Capture extra headers

OCPI headers pass through an allowlist, so Authorization and everything else unlisted is dropped at capture. ocpiAllowedHeaders extends the list and can never shrink it.

evpanda.StartOCPI(evpanda.OCPIConfig{
    OCPIAllowedHeaders: []string{"x-trace-id"},
})

startOCPI({ ocpiAllowedHeaders: ["x-trace-id"] });

evpanda.start_ocpi(evpanda.OCPIConfig(ocpi_allowed_headers=["x-trace-id"]))

See Data flow for the defaults.

Common mistakes

Mistake	Cost
Auth inside the capture middleware (Go)	Everything drops, on every request, while the requests themselves work
Capturing only inbound traffic	You see your partners’ failures but never your own
Sending your own platform as the identity	Per-partner error rates become meaningless
No body parser (Node express)	Exchanges captured with no request body
Wrapping fetch but calling axios	Nothing captured outbound
Not closing response bodies	Outbound messages that never ship

Data flow

Exchange shape, the header allowlist, credentials masking.

Getting started

Config, stats, shutdown, troubleshooting.

Core concepts

SDK integration

Model

Adapters

Telling the adapters who the partner is

The X-EVPanda headers

Custom resolvers

Complete example

Per-SDK notes

Capturing without an adapter

Capture extra headers

Common mistakes

Next

Data flow

Getting started

​Model

​Adapters

​Telling the adapters who the partner is

​The X-EVPanda headers

​Custom resolvers

​Complete example

​Per-SDK notes

​Capturing without an adapter

​Capture extra headers

​Common mistakes

​Next

Data flow

Getting started

Model

Adapters

Telling the adapters who the partner is

The X-EVPanda headers

Custom resolvers

Complete example

Per-SDK notes

Capturing without an adapter

Capture extra headers

Common mistakes

Next