Python LangChain Over Unix Domain Sockets

This guide starts from the same response-resumption checkpoint as Python Response Recording and Resumption. The only change is transport configuration.

Local Memory Service

Note: The memory-service CLI has not yet been released as a standalone binary. You can install it from source with Go:
go install -tags "sqlite_fts5 sqlite_json" github.com/chirino/memory-service@latest

memory-service serve \
  --db-kind=sqlite \
  --db-url=file:$HOME/.local/share/memory-service/memory.db \
  --vector-kind=sqlite \
  --cache-kind=local \
  --unix-socket=$HOME/.local/run/memory-service/api.sock

Those options are chosen to keep the example fully local: SQLite backs both the datastore and vector search, cache-kind=local avoids any external cache dependency, the database file lives under $HOME/.local/share, and the socket lives under $HOME/.local/run so filesystem permissions in the user’s home directory help protect access.

Agent Configuration

No Python source file changes are required. This checkpoint already builds the checkpointer, history middleware, proxy, and response recorder from package defaults:

app.py

checkpointer = MemoryServiceCheckpointSaver.from_env()
history_middleware = MemoryServiceHistoryMiddleware.from_env()

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer,
    middleware=[history_middleware],
    system_prompt="You are a Python memory-service demo agent.",
)

app = FastAPI(title="Python LangChain Agent With Response Recording and Resumption")


@app.get("/ready")
async def ready() -> dict[str, str]:
    return {"status": "ok"}
install_fastapi_authorization_middleware(app)
proxy = MemoryServiceProxy.from_env()
recording_manager = MemoryServiceResponseRecordingManager.from_env()

That means the file you update is the one that sets environment variables before the app starts. If you keep those in a .env file next to the app, this .env.example snippet shows the exact value to add:

.env.example

MEMORY_SERVICE_UNIX_SOCKET=$HOME/.local/run/memory-service/api.sock

MemoryServiceCheckpointSaver, MemoryServiceHistoryMiddleware, MemoryServiceProxy, and MemoryServiceResponseRecordingManager will route both REST and gRPC over that socket automatically once that environment variable is present.

Define the bearer-token helper used by the checkpoint app:

function get-token() {
  curl -sSfX POST http://localhost:8081/realms/memory-service/protocol/openid-connect/token \
    -H "Content-Type: application/x-www-form-urlencoded" \
    -d "client_id=memory-service-client" \
    -d "client_secret=change-me" \
    -d "grant_type=password" \
    -d "username=bob" \
    -d "password=bob" \
    | jq -r '.access_token'
}

Start a streaming response, but disconnect before it finishes. With a short timeout, the client may disconnect before the first visible SSE chunk arrives; the next step verifies that the response is still resumable:

curl -NsSfX POST http://localhost:9090/chat/9b3c1d74-53c2-4df0-a680-7a63d4550d91 \
  --max-time 0.2 \
  -H "Content-Type: text/plain" \
  -H "Authorization: Bearer $(get-token)" \
  -d "Write a short story about a cat."

Check that the response is resumable:

curl -sSfX POST http://localhost:9090/v1/conversations/resume-check \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $(get-token)" \
  -d '["9b3c1d74-53c2-4df0-a680-7a63d4550d91"]'

Example output:

[
  "9b3c1d74-53c2-4df0-a680-7a63d4550d91"
]

Resume the interrupted response:

curl -NsSfX GET http://localhost:9090/v1/conversations/9b3c1d74-53c2-4df0-a680-7a63d4550d91/resume \
  --max-time 0.5 \
  -H "Authorization: Bearer $(get-token)"

Example output:

data: {"eventType":"PartialResponse","chunk":"Once "}

Verify that proxied REST traffic also works:

curl -sSfX GET http://localhost:9090/v1/conversations/9b3c1d74-53c2-4df0-a680-7a63d4550d91 \
  -H "Authorization: Bearer $(get-token)"

Example output:

{
  "id": "9b3c1d74-53c2-4df0-a680-7a63d4550d91"
}