netbird/docs/agent-networks/modules/32-proxy-llm-parsers.md

proxy/llm-parsers — SDK adapters + pricing + SSE

The runtime-agnostic LLM library: the OpenAI Responses API (/v1/responses) and the older Chat Completions API (/v1/chat/completions), the Anthropic Messages API (/v1/messages), the SSE wire format (event: / data: lines, \n\n framing, CRLF tolerance), and per-provider token accounting (OpenAI's cached-prompt subset vs Anthropic's cache_read additive model). The pricing table's per-provider cost formula is the highest-leverage place a small bug would silently mis-bill operators.

Sibling module: 31-proxy-middleware-builtin.md — the 8 middlewares that consume this package's parsers + pricing table.

---

Module boundary

proxy/internal/llm is the runtime-agnostic LLM library shared by every middleware that needs to understand provider-specific shapes. Zero proxy-framework dependencies:

• parser.go — Parser interface, Provider enum, public factories (Parsers, DetectParser, ParserByName).
• openai.go / anthropic.go / bedrock.go — per-provider Parser impls.
• sse.go — SSE scanner (Scanner, Event, NewScanner).
• errors.go — sentinels callers branch on with errors.Is.
• pricing/ — immutable pricing table + the per-surface cost formula. The rates themselves come from management inside cost_meter's middleware config; this package holds no price list and reads no files.
• fixtures/ — captured request/response/stream bodies the tests replay.

The package carries zero proxy-framework dependencies so the same parsers can be reused later by a WASM adapter (parser.go:1–6).

Files

File	LOC	Notes
parser.go	104	Interface + factories + Provider{Unknown,OpenAI,Anthropic} enum
openai.go	347	Chat Completions + Completions + Responses API; cached_tokens subset
openai_test.go	222	11 tests; fixture replay + cached/Responses-API matrix
anthropic.go	172	Messages + legacy /v1/complete; cache_read + cache_creation additive
anthropic_test.go	154	7 tests including streaming-extraction-skipped contract
bedrock.go	190	AWS Bedrock InvokeModel (snake_case) + Converse (camelCase) response shapes; model lives in URL path
bedrock_test.go	—	InvokeModel + Converse usage shapes; AWS event-stream content-type → ErrStreamingUnsupported on buffered ParseResponse
sse.go	117	bufio-backed scanner; CRLF normalised; trailing-event handling
sse_test.go	175	12 tests; fixture replay + multiline + size limits
parser_test.go	53	Parsers(), DetectParser, provider enum values
errors.go	31	6 sentinels: Err{Unknown,Unsupported}Provider/Model, Err{NotLLM,Malformed}Response, ErrStreamingUnsupported, ErrMalformedRequest
pricing/pricing.go	234	Table, Entry, EntryJSON, Costs; NewTable/NewEntries validation + EntryCosts formula. No I/O, no reload, no embedded rates
pricing/pricing_test.go	177	10 tests — provider-shape formulas, cached clamp, rate fallback, nil-safety, rate validation
fixtures/*	21–59	OAI chat/responses/stream + Anthro messages/stream

Request body → parser dispatch

flowchart TD
    A[HTTP request<br/>URL + JSON body] --> B{ParserByName?<br/>provider_id config set}
    B -- yes --> P[matched Parser]
    B -- no --> C[DetectParser]
    C --> D{loop Parsers<br/>OpenAIParser, AnthropicParser}
    D -- DetectFromURL match --> P
    D -- no match --> X[ok=false<br/>middleware skips]
    P --> E[ParseRequest body]
    E -->|err: ErrMalformedRequest| Y[middleware emits provider only]
    E --> F[RequestFacts<br/>model + stream]
    P --> G[ExtractPrompt body]
    G --> H[joinMessages<br/>extractContentParts<br/>decodeStringOrJoin]
    H --> I[prompt text<br/>or empty]
    F --> J[stamps llm.model + llm.stream]
    I --> K[stamps llm.request_prompt_raw<br/>subject to capture_prompt gate]

OpenAI's URL hints (openai.go:27–33) include both /v1/chat/completions and the bare /chat/completions — the latter covers Cloudflare AI Gateway, which rewrites the canonical version segment. Anthropic's hints are /v1/messages and /v1/complete (anthropic.go:14–17). Both implementations use case-insensitive substring matching so a proxy prefix strip / rewrite doesn't defeat detection.

ParserByName (parser.go:93–103) is the agent-network bypass: the synthesiser knows which parser to use because it built the synth service from the catalog, so it stamps provider_id on the parser config and the middleware skips URL sniffing entirely. This is what makes the same parser set work whether the request flows to OpenAI direct, to LiteLLM, to Portkey, or to any gateway with a non-canonical URL shape.

Path-routed providers (Vertex AI, Bedrock) bypass both ParserByName and DetectParser. The model and the parser surface live in the URL path, so the request middleware extracts them directly (parseVertexPath / parseBedrockPath) before the parser-selection step. For Vertex the publisher segment picks the parser (anthropic → Anthropic parser; google/Gemini → none, request denied as unmeterable). For Bedrock the dedicated BedrockParser handles the response. Full treatment in 50-path-routed-providers.md.

Streaming response → SSE chunker → response parser → completion + token count

sequenceDiagram
    participant U as upstream LLM
    participant LR as llm_response_parser<br/>(OnResponse)
    participant S as llm.NewScanner<br/>(SSE framer)
    participant P as Parser-specific accumulator<br/>(accumulateOpenAIStream<br/>or accumulateAnthropicStream)

    U-->>LR: text/event-stream<br/>(buffered prefix in RespBody)
    LR->>S: NewScanner(bytes.NewReader(body))
    loop until EOF or [DONE]
        S-->>LR: Event{Type, Data}
        LR->>P: dispatch per event.Type<br/>(OpenAI: data-only<br/>Anthropic: named events)
        P-->>P: accumulate completion text<br/>track usage from final frame
    end
    P-->>LR: llm.Usage + completion string
    LR->>LR: appendUsage stamps<br/>llm.{input,output,total,cached_input,cache_creation}_tokens
    LR->>LR: truncateCompletion(3500 bytes, rune-safe)
    LR->>LR: redactPII if redact_pii && captureCompletion

Scanner.Next (sse.go:44–87) returns one event per \n\n boundary; multiple data: lines join with \n; comment lines (starting with :) are skipped per the SSE spec; a trailing event without a closing blank line is still returned before io.EOF so a server that closes the connection cleanly doesn't lose the last frame (sse.go:55–58). CRLF is normalised in trimEOL so fixtures captured from live servers replay unchanged.

Per-provider

OpenAI

openai.go:54–67 defines openAIRequest with three prompt fields: messages (Chat Completions), prompt (legacy), input (Responses API). The decoder uses json.RawMessage so each shape is parsed lazily.

ParseResponse (openai.go:117–146) accepts both naming conventions: Chat Completions returns prompt_tokens/completion_tokens, Responses API returns input_tokens/output_tokens. pickInt64 prefers Responses-API names and falls back — same parser handles both endpoints without per-route config. openAICachedTokens mirrors the fallback for input_tokens_details.cached_tokens vs prompt_tokens_details.cached_tokens.

Key invariant: CachedInputTokens for OpenAI is a SUBSET of InputTokens. The cost meter clamps to guard against malformed upstream responses where cached > total.

Anthropic

anthropic.go:37–49 defines anthropicRequest covering Messages API (system + messages[]) and legacy /v1/complete (prompt string). ExtractPrompt emits system: <text> first when present, then per-message role: content.

ParseResponse (anthropic.go:82–104) fills three independent token buckets: InputTokens, CacheReadInputTokens, CacheCreationInputTokens. Latter two are additive (not subset). TotalTokens sums all four so downstream dashboards render one "tokens" number without double-counting.

ExtractCompletion walks content[] {type, text} parts and concatenates non-empty text with newlines, falling back to legacy completion.

Bedrock

bedrock.go implements the Parser interface for the AWS Bedrock runtime. Bedrock is path-routed: the model lives in the URL (/model/{id}/{action}), so the request middleware extracts it (see 50-path-routed-providers.md) and ParseRequest is a deliberate no-op. The parser's real work is on the response leg, covering both Bedrock body shapes:

• InvokeModel — vendor-native. Anthropic-on-Bedrock returns snake_case usage (input_tokens, output_tokens, cache_read_input_tokens, cache_creation_input_tokens) with the same additive cache buckets as first-party Anthropic.
• Converse — unified camelCase (inputTokens, outputTokens, totalTokens). firstNonZero folds the two naming conventions into one Usage; when Converse omits totalTokens the parser sums the buckets.

ProviderName() returns "bedrock" — its own pricing surface in the table management ships, keyed by the normalised model id (region prefix + version suffix stripped by the request parser; management normalises its keys the same way at synth time so the two compare equal). ParseResponse returns ErrStreamingUnsupported for an AWS binary event-stream content-type (application/vnd.amazon.eventstream, isAWSEventStream) so the caller routes to the streaming accumulator instead.

SSE framing

Scanner is bufio-backed, 64 KiB read buffer, 1 MiB max line so a malicious upstream can't blow process memory (sse.go:33–38, 97–100). splitField strips one space after the : per the SSE spec. Documented not safe for concurrent use; every consumer creates a fresh scanner per response body. Streaming accumulators live in the middleware package (llm_response_parser/streaming.go) but use llm.NewScanner so the framing contract stays here.

Pricing table

Management is the sole pricing authority. The proxy carries no embedded price list and reads no pricing file: the whole table arrives inside cost_meter's ConfigJSON on the ordinary mapping push, and a price change is just another push — the chain rebuild constructs a fresh Table, so there is nothing to reload (pricing.go:1–7). The management side of the contract (catalog defaults, the operator's stored per-provider prices, and AgentNetwork.PricingDefaultsFile) is covered in the management-side module guide; cost_meter's wire shape is in 31-proxy-middleware-builtin.md.

EntryJSON (pricing.go:36–45) is the management→proxy contract — five USD-per-1k rates under input_per_1k, output_per_1k, cached_input_per_1k, cache_read_per_1k, cache_creation_per_1k. Management's pricing.Entry marshals the identical names, and EntryJSON/Entry are field-identical so NewEntries converts by direct struct conversion rather than field-by-field copying (a new rate can't be silently dropped in transit).

EntryCosts (pricing.go:183–234) is the cost formula — most security-relevant math in this module. The surface (the llm.provider value the request parser stamped) selects the formula, never the tier the entry came from: a per-provider-record override on an Anthropic route still bills its cache buckets additively.

Provider	Formula
openai	(inTokens − clamped) × InputPer1K + clamped × CachedInputPer1K + outTokens × OutputPer1K where clamped = min(cachedInput, inTokens)
anthropic, bedrock	inTokens × InputPer1K + cachedInput × CacheReadPer1K + cacheCreation × CacheCreationPer1K + outTokens × OutputPer1K
default	inTokens × InputPer1K + outTokens × OutputPer1K

bedrock shares the Anthropic additive-cache formula (pricing.go:214–229): Anthropic-on-Bedrock reports the same additive cache buckets, while non-Anthropic Bedrock models (Nova, Llama) simply report zero in those buckets so cost reduces to input + output.

Each per-bucket rate falls back to InputPer1K when zero — operators opt in to discounts by setting the field.

Costs (pricing.go:143–163) is the per-request split. The four per-bucket fields are the base; TotalUSD and CacheUSD are derived in newCosts so the aggregates can never drift from the breakdown. InputUSD is always the non-cached input bucket on both provider shapes, so input and cached-input never double-count.

Public contracts

Parser interface (parser.go:50–66):

type Parser interface {
    Provider() Provider
    ProviderName() string
    DetectFromURL(path string) bool
    ParseRequest(body []byte) (RequestFacts, error)
    ParseResponse(status int, contentType string, body []byte) (Usage, error)
    ExtractPrompt(body []byte) string
    ExtractCompletion(status int, contentType string, body []byte) string
}

Adding a provider means implementing this interface and appending to the slice returned by Parsers() (parser.go:78–84). Order matters: DetectFromURL ties resolve by registration order. Parsers() today returns {OpenAIParser, AnthropicParser, BedrockParser}.

Provider enum (parser.go:8–18): ProviderUnknown = 0, ProviderOpenAI = 1, ProviderAnthropic = 2, ProviderBedrock = 3. Numeric values are persisted in nothing today but treat them as wire-stable — new providers must take fresh numbers.

Pricing construction + lookup (pricing.go:60–130):

func NewEntries(raw map[string]map[string]EntryJSON) (map[string]map[string]Entry, error)
func NewTable(raw map[string]map[string]EntryJSON) (*Table, error)

func (t *Table) Lookup(provider, model string) (Entry, bool)
func (t *Table) Cost(provider, model string, inTokens, outTokens, cachedInput, cacheCreation int64) (float64, bool)
func (t *Table) Costs(provider, model string, inTokens, outTokens, cachedInput, cacheCreation int64) (Costs, bool)
func EntryCosts(entry Entry, surface string, inTokens, outTokens, cachedInput, cacheCreation int64) Costs

NewTable is the surface-keyed defaults table; NewEntries returns the raw two-level map cost_meter uses for the per-provider-record tier (it looks up an Entry directly and calls EntryCosts, so it needs no Table wrapper). Both reject any non-finite or negative rate, so a corrupt config fails the chain build rather than mispricing silently. Nil input yields an empty, never-matching table.

Nil-safe: t.Cost/t.Lookup on a nil receiver returns ok=false (pricing.go:96–99). ok=false means the surface or model is absent from the table management sent; the caller emits cost.skipped=unknown_model.

Invariants

1. The pricing package is pure and platform-independent. No file I/O, no //go:embed, no goroutines, no build tags — the rates arrive as config, so there is nothing platform-specific left to port. Anything reintroducing a read-from-disk path here re-splits pricing authority between management and the proxy, which is exactly what this design removed.

2. SSE scanner handles partial chunks. A buffered prefix that doesn't end in \n\n still yields its accumulated event before io.EOF (sse.go:55–58). Tests: TestSSEScanner_OpenAIFixture, TestSSEScanner_AnthropicFixture, TestSSEScanner_MultilineData, TestSSEScanner_CRLF. The streaming accumulators ride on this: accumulateAnthropicStream and accumulateOpenAIStream break on any scanner error to return partial usage rather than aborting (streaming.go:68–73, 144–150).

3. Management is the only source of rates. Table has no constructor that invents prices: the only way in is NewTable/NewEntries over the wire map management sent. A missing or empty pricing block therefore means no prices at all (cost_meter records cost.skipped=unknown_model, $0) — never a stale built-in fallback that would silently bill list price.

4. Tables are immutable once built. Table.entries is written only in NewEntries and never mutated afterwards, and cost_meter's perRecord map is likewise build-time-only (pricing.go:47–52). This is what makes the no-reload design safe: a price change arrives as a mapping push that builds a new middleware instance over a new table, so concurrent readers can't observe a half-updated price list and no atomic swap or lock is needed on the hot path.

5. Rate validation happens at chain-build time, not per request. NewEntries rejects negative, NaN, and ±Inf rates field by field (pricing.go:60–83), naming the offending surface/model/field in the error. Management enforces the same constraints at its API boundary and in its YAML parser, so this is defense-in-depth — but it means a corrupt push fails loudly at build instead of producing negative costs on live traffic. Test: TestNewTable_ValidatesRates.

6. New rates must be added to Entry, EntryJSON, and management's pricing.Entry together. NewEntries converts by direct struct conversion Entry(e) (pricing.go:76–78), which only compiles while the two structs stay field-identical — so the proxy half is compiler-enforced. The management half is not: a rate added there but not here unmarshals into nothing and prices that bucket at InputPer1K.

Things to scrutinise

Correctness. Verify the OpenAI cached-prompt clamp at pricing.go:203–206 short-circuits before subtraction. Negative token counts are clamped to zero up front (pricing.go:186–197) so no formula can yield a negative cost. Anthropic.TotalTokens sums all four buckets (in + out + cache_read + cache_creation) — downstream dashboards need to know this differs from input + output. OpenAIParser.ExtractPrompt falls through messages → input → prompt; a request sending all three reports only messages (uncommon but worth noting).

Security. Scanner.maxLine = 1 MiB; a 2 MiB single-line data: event errors from Scanner.Next and both accumulators stop with partial usage. Pricing is no longer file-backed, so the loader's path-traversal / symlink / oversize surface is gone entirely — the config channel (an authenticated mapping push from management) is now the only way rates enter the proxy, and NewEntries is the validation boundary on it. A new rate added to management's pricing.Entry but not to EntryJSON here is the remaining silent-mispricing path (see invariant 6).

Concurrency. Nothing in this package is shared mutable state: tables are built once and never written again, so cost_meter's hot path is lock-free by construction rather than by atomic swap. Per-call Scanner instances mean no shared state across concurrent response-parser calls.

Perf. Table.Cost is two map lookups + multiplications, O(1); the per-provider-record tier adds at most one more lookup. Scanner.Next is one ReadString('\n') per line. No background goroutines and no per-request allocation of pricing state.

Observability. A config carrying no pricing block logs one warning at chain-build time (cost_meter factory) and then records cost.skipped=unknown_model per request, so an old-management deployment is visible in both logs and the access log rather than quietly reporting $0. Parser errors return sentinels — middleware uses errors.Is to map to the right cost.skipped reason.

Test coverage

File	Tests	Coverage highlights
parser_test.go	3	Parsers() shape lock, DetectParser URL matrix, provider enum stability
openai_test.go	11	Chat Completions + Responses API + legacy prompt; cached-tokens subset for both naming conventions; fixture replays
anthropic_test.go	7	Messages + legacy /v1/complete; streaming REJECTED on ParseResponse (must use scanner); fixture replays
sse_test.go	12	Fixture replay both providers; multiline data:; CRLF; comment skip; trailing-event-without-blank-line; oversize rejection
pricing/pricing_test.go	10	Provider-shape switch (surface selects the formula); cached-rate + cache-read/creation fallback to InputPer1K; cached-clamp; negative-token clamp; nil-receiver safety; rate validation (negative / NaN / Inf rejected); nil + empty table

Fixtures (proxy/internal/llm/fixtures/): openai_chat_completion.json (chat.completions with usage), openai_responses.json (Responses API shape), openai_stream.txt (3 deltas + usage + [DONE]), anthropic_messages.json (Messages API non-streaming), anthropic_stream.txt (full 7-event sequence: message_start → content_block_{start,delta×2,stop} → message_delta (usage) → message_stop). No pricing fixture: the table is config-delivered, so pricing tests construct it in-process from a wire-shape map.

Cross-references

• Sibling: 31-proxy-middleware-builtin.md — the chain that calls llm.Parsers(), llm.ParserByName, llm.NewScanner, pricing.NewTable / pricing.NewEntries.
• Path-routed providers (Vertex AI + Bedrock), credential syntax, and the Bedrock AWS event-stream accumulator: 50-path-routed-providers.md.
• Direct callers: llm_request_parser/middleware.go:82–94, llm_response_parser/middleware.go:113–123, llm_response_parser/streaming.go:65, 142, cost_meter/factory.go:49–57.
• Related elsewhere: the agent-network synthesiser stamping provider_id is covered in the management-side module guide; proxy server boot + FactoryContext construction is covered in the proxy-framework guide.