Skip to content

Integrations

pytest plugin: agent suites as pytest tests (FR-S6).

Installed with AgentLens and loaded automatically by pytest (entry point pytest11: agentlens); it does nothing unless its options or fixture are used.

Suite files. Every case of a suite becomes one test item:

pytest --agentlens suites/support.yaml --agentlens suites/billing.yaml

collects suites/support.yaml::<case_id> items next to the normal tests (select them with -k, by case id, suite name or tag). The agent runs every repetition of the selected cases (one experiment per suite, honouring the suite's concurrency), the case's evaluators score each run, and a case passes when every repetition passed. A failure lists, per failing repetition, the agent error and each failed evaluator with its score and reason. Options: --agentlens-repetitions N overrides the suites' repetitions, --agentlens-db PATH saves the experiments (traces and scores) to a SQLite database, e.g. for agentlens compare.

Python tests. The agentlens_run fixture runs an agent on one case:

from agentlens.integrations.pytest_plugin import assert_passed

def test_refund(agentlens_run):
    result = agentlens_run(my_agent, {"id": "refund", "input": "refund 1234",
                                      "expected": "Refunded"},
                           evaluators=["no_error", "contains"])
    assert_passed(result)                 # AssertionError listing evaluator reasons
    assert result.mean_scores()["contains"] == 1.0

CaseFailedError

CaseFailedError(result: CaseResult, suite: str | None = None)

Bases: AssertionError

A case had at least one failing repetition; str() is the readable report.

SuiteRunError

Bases: Exception

A suite could not run at all (agent import failure, bad evaluator spec, store error).

AgentlensSuiteFile

Bases: File

A suite YAML file: loads the suite and yields one AgentlensCase per case.

The suite runs once, when its first selected case runs: only the selected cases (after -k/-m/node-id selection), as one experiment.

collect

collect() -> Iterator[AgentlensCase]

Load the suite (a SuiteError is a collection error).

repr_failure

repr_failure(excinfo: ExceptionInfo[BaseException]) -> str | TerminalRepr

Suite errors are shown as their file:line:column issues, without a traceback.

experiment_run

experiment_run() -> ExperimentRun

The suite's run, executing the selected cases on first call.

Raises:

Type Description
SuiteRunError

if the suite could not run.

AgentlensCase

AgentlensCase(*, case: Case, **kwargs: Any)

Bases: Item

One case of a suite file (all its repetitions).

suite_file property

suite_file: AgentlensSuiteFile

The suite file this case belongs to.

runtest

runtest() -> None

Run (or reuse) the suite's experiment and fail unless every repetition passed.

repr_failure

repr_failure(excinfo: ExceptionInfo[BaseException], style: Any = None) -> str | TerminalRepr

The evaluator report for failed cases, the reason for suites that could not run.

reportinfo

reportinfo() -> tuple[Path, int | None, str]

(suite file, None, "agentlens <suite>::<case>") for report headers.

CaseRunner

CaseRunner(*, name: str, store: SQLiteStore | None, repetitions: int | None)

What the agentlens_run fixture returns: call it to run an agent on one case.

format_case_result

format_case_result(result: CaseResult, *, suite: str | None = None) -> str

A readable report of result: the case, then every failing repetition with its error or output and each failed evaluator (name: score S: reason).

Passing repetitions are only counted. Used for failed suite items and by assert_passed.

assert_passed

assert_passed(result: CaseResult, *, suite: str | None = None) -> None

Fail the current test unless every repetition of result passed.

Raises:

Type Description
CaseFailedError

an AssertionError whose message is format_case_result (evaluator, score and reason of each failure).

pytest_addoption

pytest_addoption(parser: Parser) -> None

Register --agentlens, --agentlens-db and --agentlens-repetitions.

pytest_configure

pytest_configure(config: Config) -> None

Validate the options and add the suite files to the paths pytest collects.

pytest_unconfigure

pytest_unconfigure(config: Config) -> None

Close the --agentlens-db store, if it was opened.

pytest_collect_file

pytest_collect_file(file_path: Path, parent: Collector) -> AgentlensSuiteFile | None

Collect the files passed with --agentlens (and nothing else).

agentlens_run

agentlens_run(request: FixtureRequest) -> CaseRunner

Run an agent on one case and evaluate it: agentlens_run(agent, case, evaluators=...).

Returns a CaseRunner; see assert_passed for a readable failure.

Post a report as a pull-request comment, updating it in place (FR-C3).

The GitHub Action (action.yml) calls agentlens github comment --file report.md, which runs upsert_pr_comment:

  1. list the PR's comments (GET /repos/{owner}/{repo}/issues/{n}/comments, every page);
  2. if one contains the hidden marker (<!-- agentlens-report -->), edit it (PATCH /repos/{owner}/{repo}/issues/comments/{id}), unless its body is already the same;
  3. otherwise create one (POST .../issues/{n}/comments).

So a PR gets one AgentLens comment that follows its latest push instead of a new comment per push. GitHubContext.from_env reads what a workflow provides: GITHUB_TOKEN (pass secrets.GITHUB_TOKEN or the action's github-token), GITHUB_REPOSITORY (owner/repo), GITHUB_EVENT_PATH (the event JSON, for the PR number) and GITHUB_API_URL (GitHub Enterprise).

The token needs pull-requests: write (or issues: write). Pull requests from forks get a read-only token on the pull_request event, so posting fails there with a GitHubError saying so.

MAX_COMMENT_CHARS module-attribute

MAX_COMMENT_CHARS = 65536

GitHub rejects longer comment bodies; longer reports are truncated with a note.

GitHubError

Bases: RuntimeError

Missing configuration, or the GitHub API refused or failed a request.

CommentResult dataclass

CommentResult(action: Literal['created', 'updated', 'unchanged'], comment_id: int, url: str)

What upsert_pr_comment did.

url instance-attribute

url: str

The comment's html_url (empty if the API did not return one).

GitHubContext dataclass

GitHubContext(repository: str, pr_number: int, token: str, api_url: str = DEFAULT_API_URL)

Where to comment and with which token. repr hides the token.

repository instance-attribute

repository: str

owner/repo.

from_env classmethod

from_env(env: Mapping[str, str] | None = None, *, repository: str | None = None, pr_number: int | None = None, token: str | None = None) -> GitHubContext

The context of the running workflow; explicit arguments win over env.

env defaults to os.environ. The PR number comes from the event file: pull_request.number (pull_request / pull_request_target events), issue.number (comments) or number.

Raises:

Type Description
GitHubError

if the token, the repository or the PR number is missing.

with_marker

with_marker(body: str, marker: str = DEFAULT_MARKER) -> str

body starting with <!-- marker --> (added only if it is not in the body).

upsert_pr_comment

upsert_pr_comment(body: str, *, context: GitHubContext, marker: str = DEFAULT_MARKER, client: Client | None = None) -> CommentResult

Create the PR's report comment, or update the one carrying marker.

Parameters:

Name Type Description Default
body str

The comment (e.g. render(report, "markdown")). The marker is added if missing; a body longer than MAX_COMMENT_CHARS is truncated.

required
context GitHubContext

Repository, PR number and token (GitHubContext.from_env).

required
marker str

Key of the hidden <!-- marker --> identifying the comment; use one per report when a PR gets several.

DEFAULT_MARKER
client Client | None

An httpx.Client to use (tests, proxies); by default one is created and closed.

None

Returns:

Type Description
CommentResult

Whether the comment was created, updated or already unchanged,

CommentResult

with its id and URL. If several comments carry the marker, the oldest

CommentResult

is updated.

Raises:

Type Description
GitHubError

if a request fails or GitHub answers with an error.

ValueError

if marker is not a valid key.

Export AgentLens traces as OpenTelemetry spans (FR-T6), following the GenAI conventions.

Install the extra (pip install "agentlens-toolkit[otel]"), configure OpenTelemetry as usual (an OTLP exporter to Jaeger, Grafana Tempo, Langfuse, Phoenix, ...) and pass an OTelExporter as the trace sink, or call export_trace on traces you already have:

import agentlens
from agentlens.integrations.otel import OTelExporter, fan_out
from agentlens.store import SQLiteStore

exporter = OTelExporter()                 # global TracerProvider by default
with SQLiteStore() as store, agentlens.run(
    "support", sink=fan_out(store.save_trace, exporter)
):
    ...

Mapping (OpenTelemetry semantic conventions for generative AI, gen_ai.*):

===================== ========================== ======== ================================= AgentLens OTel span name kind main attributes ===================== ========================== ======== ================================= the trace (root) invoke_agent {name} INTERNAL gen_ai.operation.name, gen_ai.agent.name agent_step invoke_agent {name} INTERNAL same llm_call chat {model} CLIENT gen_ai.provider.name, gen_ai.request.model, gen_ai.request.temperature, gen_ai.request.max_tokens, gen_ai.response.model, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens tool_call execute_tool {tool} INTERNAL gen_ai.tool.name retrieval/custom the span name INTERNAL agentlens.span.kind ===================== ========================== ======== =================================

Every span also carries agentlens.trace_id / agentlens.span_id (to find it in the store) and, when known, agentlens.usage.cost_usd. Errors set the span status to ERROR and error.type to the exception type. Start and end times are the recorded ones, so the exported timeline matches the trace.

Content is not exported by default. Prompts, outputs and tool arguments may hold personal data; capture_content=True adds them (gen_ai.input.messages, gen_ai.output.messages, gen_ai.tool.call.arguments, gen_ai.tool.call.result, as JSON strings), the opt-in the conventions recommend. Use the redacted trace the sink receives, never the live one.

INSTRUMENTATION_NAME module-attribute

INSTRUMENTATION_NAME = 'agentlens'

The instrumentation scope of the exported spans.

OTelExporter

OTelExporter(*, tracer_provider: TracerProvider | None = None, capture_content: bool = False)

A trace sink that exports every finished trace to OpenTelemetry.

agentlens.run(..., sink=OTelExporter()); combine it with the store with fan_out. Export errors are logged, never raised into the agent (FR-T5).

span_name

span_name(span: Span) -> str

The OTel span name: "{operation} {target}" for GenAI spans, else the name.

span_attributes

span_attributes(span: Span, *, capture_content: bool = False) -> dict[str, AttributeValue]

The OTel attributes of one AgentLens span (see the module docstring).

export_trace

export_trace(trace: Trace, *, tracer: Tracer | None = None, tracer_provider: TracerProvider | None = None, capture_content: bool = False, context: Context | None = None) -> int

Emit trace as OpenTelemetry spans (one root span plus one per AgentLens span).

Parameters:

Name Type Description Default
trace Trace

A finished trace (unfinished spans end with the trace).

required
tracer Tracer | None

The tracer to use; by default one named "agentlens" from tracer_provider (else the global provider).

None
tracer_provider TracerProvider | None

Where to get the tracer from when tracer is not given.

None
capture_content bool

Also export prompts, outputs and tool arguments/results.

False
context Context | None

Parent context of the root span; by default a new trace (the agent run is exported after the fact, so the ambient span is not its parent).

None

Returns:

Type Description
int

The number of OTel spans emitted.

fan_out

fan_out(*sinks: Callable[[Trace], Any]) -> Callable[[Trace], None]

One sink calling every sink in order; one failing does not stop the others.

agentlens.run(name, sink=fan_out(store.save_trace, OTelExporter())).