Python Instrumentation

from lmnr import Laminar, Instruments

# Enable only specific instruments
Laminar.initialize(
    instruments={Instruments.OPENAI, Instruments.ANTHROPIC}
)

# Disable specific instruments
Laminar.initialize(
    disabled_instruments={Instruments.LANGCHAIN}
)

import asyncio

from daytona import AsyncDaytona, SessionExecuteRequest
from lmnr import Laminar, observe

Laminar.initialize()

@observe()
async def run_sandbox_command():
    daytona = AsyncDaytona()
    sandbox = await daytona.create()

    session_id = "streaming-session"
    await sandbox.process.create_session(session_id)

    await sandbox.process.execute_session_command(
        session_id,
        SessionExecuteRequest(
            command=(
                'for i in {1..5}; do echo "Processing iteration $i of 5 - '
                'This is a longer test message to verify streaming output handling '
                'with substantial text content"; '
                'echo "Error at iteration $i: This is a simulated error message '
                'with extended details for testing purposes" >&2; sleep 1; done'
            ),
            run_async=True,
        ),
    )

    print("Continuing execution while command runs in background...")
    await asyncio.sleep(10)
    print("Other operations completed!")

    await sandbox.delete()

if __name__ == "__main__":
    asyncio.run(run_sandbox_command())

SELECT time, severity_text, body, attributes
FROM logs
WHERE time > now() - INTERVAL 1 DAY
ORDER BY time DESC

from lmnr import SessionRecordingOptions, MaskInputOptions

options = SessionRecordingOptions(
    mask_input_options=MaskInputOptions(...)
)

Laminar.initialize(session_recording_options=options)