Appendix: inside the local Langfuse
The server the lessons send to, a few lines at a time. Nothing here is Langfuse; skip it unless you want to know how the requests are answered.
Copy this file as local_langfuse.py beside your lesson files. It answers what the SDK sends, keeps everything it receives, and refuses to talk to any other machine.
What it keeps, and what it answers
"""A stand-in for the Langfuse server, on your own machine.
The Langfuse SDK sends observations to /api/public/otel/v1/traces and uses a
REST API for scores, prompts and datasets. This server answers those requests
on 127.0.0.1 and keeps what it receives in lists you can print. Given a model
function, it also answers like OpenAI's chat API. start() blocks every
connection to another computer for the rest of the program.
"""It opens by saying what it is for, and which requests it answers.
import json
import socket
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, unquote, urlparse
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
SPANS, SCORES, REQUESTS = [], [], [] # observations, scores, "METHOD /path" of each request
PROMPTS, DATASETS = {}, {} # prompt name -> its versions, dataset name -> dataset
MODEL = None # answers /v1/chat/completions when set
NOW = "2026-03-03T09:00:00Z"Three lists and two dictionaries hold everything the server receives, so a lesson can print them. SPANS holds observations, SCORES the scores, and REQUESTS the method and path of every request, which is how a lesson shows what the SDK asked for.
The endpoint the SDK exports to
def _value(v):
kind = v.WhichOneof("value")
if kind == "array_value":
return [_value(item) for item in v.array_value.values]
return getattr(v, kind) if kind else NoneObservations arrive in OpenTelemetry's protobuf format, where every attribute value is wrapped in a type. _value unwraps one, and unwraps a list one item at a time.
def _spans(body):
request = ExportTraceServiceRequest()
request.ParseFromString(body)
for resource in request.resource_spans:
for scope in resource.scope_spans:
for span in scope.spans:
SPANS.append({
"name": span.name, "id": span.span_id.hex(), "trace": span.trace_id.hex(),
"parent": span.parent_span_id.hex(), "scope": scope.scope.name,
"attributes": {a.key: _value(a.value) for a in span.attributes},
})
return {}This is the receiver. The opentelemetry-proto package, which came with Langfuse, turns the request body back into spans, and each one becomes a dictionary with its name, its ids, the library that recorded it and its attributes.
The REST endpoints
def _scores(batch):
for event in batch:
if event["type"] == "score-create":
SCORES.append(event["body"])
return {"successes": [{"id": e["id"], "status": 201} for e in batch], "errors": []}Scores do not travel with observations. They arrive as a group of events on an endpoint of their own, and the answer tells the SDK which ones were accepted. Lesson 23 uses them.
def _prompt(name, query, body):
versions = PROMPTS.setdefault(name, [])
if body is not None: # a new version: it takes "latest" and any labels it asks for
labels = body.get("labels", []) + ["latest"]
for old in versions:
old["labels"] = [label for label in old["labels"] if label not in labels]
versions.append({"name": name, "type": body.get("type", "text"), "prompt": body["prompt"],
"config": body.get("config") or {}, "labels": labels, "tags": [],
"version": len(versions) + 1})
return versions[-1]
if "version" in query:
return versions[int(query["version"]) - 1]
label = query.get("label", "production")
return next(v for v in versions if label in v["labels"])Prompts are kept per name as a list of versions. Saving one adds a version, which takes the latest label and any label asked for; fetching one takes a version number or a label, and production when neither is given. Lessons 18 to 22 are about this.
def _relabel(path, body):
name, _, version = path.split("/")[-3:]
versions = PROMPTS[unquote(name)]
for old in versions:
old["labels"] = [label for label in old["labels"] if label not in body["newLabels"]]
versions[int(version) - 1]["labels"] += body["newLabels"]
return versions[int(version) - 1]Releasing a prompt version means moving a label to it, which the SDK does with a PATCH request. A label lives on one version at a time, so it is taken off the others first.
def _dataset(path, query, body):
if path == "/api/public/v2/datasets":
DATASETS[body["name"]] = {"id": body["name"], "name": body["name"], "items": [],
"metadata": body.get("metadata"), "projectId": "local",
"createdAt": NOW, "updatedAt": NOW}
return DATASETS[body["name"]]
if path.startswith("/api/public/v2/datasets/"):
return DATASETS[unquote(path.rsplit("/", 1)[1])]A dataset is a name, some metadata and a list of items. Lesson 25 creates one and fetches it back.
if path == "/api/public/dataset-items" and body is not None:
dataset = DATASETS[body["datasetName"]]
item = {"id": body.get("id") or f"item-{len(dataset['items']) + 1}", "status": "ACTIVE",
"input": body.get("input"), "expectedOutput": body.get("expectedOutput"),
"metadata": body.get("metadata"), "sourceTraceId": body.get("sourceTraceId"),
"datasetId": dataset["id"], "datasetName": dataset["name"],
"createdAt": NOW, "updatedAt": NOW, "mediaReferences": []}
dataset["items"].append(item)
return itemEach item is an input with an expected output, and keeps the id the SDK generated for it. sourceTraceId is how an item remembers the ticket it came from.
if path == "/api/public/dataset-items":
items = DATASETS[query["datasetName"]]["items"]
return {"data": items, "meta": {"page": 1, "limit": 50, "totalItems": len(items), "totalPages": 1}}
return {"id": "run-item", "datasetRunId": body["runName"], "datasetRunName": body["runName"],
"datasetItemId": body["datasetItemId"], "traceId": body["traceId"],
"observationId": body.get("observationId"), "createdAt": NOW, "updatedAt": NOW}Items come back a page at a time, which is how the SDK fetches a dataset in lesson 25. Anything else with dataset in its path is lesson 27 recording one case of a dataset run.
A model, when a lesson gives it one
def _chat(body):
text, usage = MODEL(body["messages"])
return {"id": "chatcmpl-1", "object": "chat.completion", "created": 0, "model": body["model"],
"choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}}],
"usage": {"prompt_tokens": usage["input"], "completion_tokens": usage["output"],
"total_tokens": usage["input"] + usage["output"]}}Given a model function, the server answers like OpenAI's chat API: the same shape of answer, with usage counts. Lesson 7 hands it the model you write in lesson 6, so an OpenAI client can be traced with no key and no network.
Sending each request to the right place
class Handler(BaseHTTPRequestHandler):
def _handle(self):
url = urlparse(self.path)
query = {key: values[0] for key, values in parse_qs(url.query).items()}
raw = self.rfile.read(int(self.headers.get("Content-Length") or 0))
body = json.loads(raw) if raw and "json" in self.headers.get("Content-Type", "") else None
REQUESTS.append(f"{self.command} {url.path}")
try:
if url.path.endswith("/otel/v1/traces"):
return self._reply(200, _spans(raw))
if url.path == "/api/public/ingestion":
return self._reply(207, _scores(body["batch"]))Every request lands here. The path, the query and the body are read once, the request is recorded in REQUESTS, and the traces endpoint is checked first because it is the busiest.
if url.path == "/api/public/projects":
return self._reply(200, {"data": [{"id": "local", "name": "shop-desk", "metadata": {},
"organization": {"id": "local", "name": "local"}}]})
if url.path.startswith("/api/public/v2/prompts/") and self.command == "PATCH":
return self._reply(200, _relabel(url.path, body))
if url.path.startswith("/api/public/v2/prompts"):
name = body["name"] if body else unquote(url.path.rsplit("/", 1)[1])
return self._reply(200, _prompt(name, query, body))
if "dataset" in url.path:
return self._reply(200, _dataset(url.path, query, body))
if url.path.endswith("/chat/completions") and MODEL:
return self._reply(200, _chat(body))The other paths in order: scores, the project list the SDK reads for auth_check, a prompt label change, prompts, datasets, and the chat endpoint when a model was given.
except (KeyError, StopIteration):
pass
self._reply(404, {"message": f"not found: {url.path}"})
do_GET = do_POST = do_PATCH = _handleA missing prompt or dataset raises KeyError or StopIteration, and both become a 404, which is what a real server would send. Every method is answered by the same function.
def _reply(self, code, data):
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(data).encode())
def log_message(self, *args):
passReplies are JSON with a status code. log_message is replaced so the server does not print a line for every request into the middle of a lesson's output.
Nothing leaves this machine
def _local_only():
"""Refuse to look up or connect to any other computer, so nothing leaves by mistake."""
lookup, connect = socket.getaddrinfo, socket.socket.connect
def guarded_lookup(host, *args, **kwargs):
if host not in ("127.0.0.1", "localhost", "::1"):
raise ConnectionRefusedError(f"blocked a connection to {host}: this lab stays on your machine")
return lookup(host, *args, **kwargs)Python looks a host name up before it connects to it. Replacing that lookup means a name other than this machine's fails before any connection is opened.
def guarded_connect(sock, address):
if isinstance(address, tuple) and address[0] not in ("127.0.0.1", "::1"):
raise ConnectionRefusedError(f"blocked a connection to {address[0]}: this lab stays on your machine")
return connect(sock, address)
socket.getaddrinfo, socket.socket.connect = guarded_lookup, guarded_connectThe same for an address that needs no lookup. From here on, any attempt to reach another computer raises, which is what the last section of this lesson relies on.
def start(model=None):
"""Start the server in the background and return its address."""
global MODEL
_local_only()
MODEL = model
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
return f"http://127.0.0.1:{server.server_address[1]}"Starting the server installs the guard first, takes the model function for the chat endpoint, and asks for port 0, which means any free port. It returns the address to give the SDK.
Reading what arrived
def tree(*fields):
"""Print the observations received so far, each under its parent, with any fields asked for."""
ids = {span["id"] for span in SPANS}
def walk(span, depth):
attributes = span["attributes"]
line = " " * depth + f"{span['name']} ({attributes.get('langfuse.observation.type', 'span')})"
for field in fields:
line += f" {field}={attributes.get('langfuse.observation.' + field)!r}"
print(line)
for child in SPANS:
if child["parent"] == span["id"]:
walk(child, depth + 1)tree prints the observations under their parents, with any observation field you name, which is how most lessons show their result.
for root in [span for span in SPANS if span["parent"] not in ids]:
walk(root, 0)A root is an observation whose parent never arrived: every trace's first observation, and, in lesson 17, a child whose parent was filtered out.
Save it as local_langfuse.py next to your lesson files. Every lesson from here on imports it.
The file ends there. It answers what the SDK sends, keeps it, and refuses any address that is not this machine, which is what lets each lesson print exactly what was exported.
- Print
local_langfuse.REQUESTSafter a lesson's run and match each path to the section above. - Add a route that answers
/api/public/healthwith{"status": "ok"}. - Break
_local_onlyby returning the real address, and watch lesson 2's last example reach the network.
You understood something today that you didn't yesterday.