API Reference¶

FastMCP server factory, run registry, event models, and configuration types used by the MCP tools and resources.

Server¶

rampa.mcp.server.build_mcp_server() ¶ function function rampa.mcp.server.build_mcp_server() ¶

Build and configure the rampa MCP server.

Returns:

FastMCP – Configured MCP server with tools and resources registered.
>>> server = build_mcp_server()
>>> server.name
’rampa’

Return type:

FastMCP

rampa.mcp.server.main() ¶ function function rampa.mcp.server.main() ¶

Entry point for the rampa MCP server.

Return type:: None

Registry¶

class rampa.mcp.registry.RunRecord ¶ class class class rampa.mcp.registry.RunRecord ¶

Bases: object

Per-run state in the registry.

>>> r = RunRecord(run_id="abc", script_path="test.py", started_at=0.0)
>>> r.is_complete
False

class rampa.mcp.registry.RuntimeRun ¶ class class class rampa.mcp.registry.RuntimeRun ¶

Bases: object

Non-serializable async state for an active run.

>>> import rampa.mcp.registry

class rampa.mcp.registry.RunRegistry ¶ class class class rampa.mcp.registry.RunRegistry ¶

Bases: object

Process-local registry of active and completed runs.

>>> reg = RunRegistry()
>>> reg.list_all()
[]

complete(run_id, result) ¶ method method complete(run_id, result) ¶

Mark a run as completed and release its runtime resources.

Parameters:

run_id (str) – The run identifier.
result (RunResult) – The final run result.

Return type:

None

Examples

>>> reg = RunRegistry()
>>> rec = RunRecord(run_id="r1", script_path="t.py", started_at=0.0)
>>> reg.register(rec)
>>> from rampa.events import RunResult, RunStatus
>>> result = RunResult(
...     run_id="r1",
...     status=RunStatus.PASSED,
...     snapshot=None,
...     threshold_results=[],
... )
>>> reg.complete("r1", result)
>>> rec.is_complete
True
>>> rec.runtime is None
True

Events¶

class rampa.events.RunResult class class class rampa.events.RunResult

Bases: object

Result of a completed test run.

>>> r = RunResult(
...     run_id="abc",
...     status=RunStatus.PASSED,
...     snapshot=None,
...     threshold_results=[],
... )
>>> r.status
<RunStatus.PASSED: 'passed'>

class rampa.events.RunStatus class class class rampa.events.RunStatus

Bases: StrEnum

Final status of a completed test run.

>>> RunStatus.PASSED.value
'passed'
>>> RunStatus.THRESHOLD_FAILED.value
'threshold_failed'

class rampa.events.PhaseEvent class class class rampa.events.PhaseEvent

Bases: EngineEvent

Lifecycle phase transition.

>>> e = PhaseEvent(run_id="abc", timestamp_ns=0, phase="setup")
>>> e.phase
'setup'

class rampa.events.SnapshotEvent class class class rampa.events.SnapshotEvent

Bases: EngineEvent

Periodic metric snapshot.

>>> from rampa.metrics import MetricSnapshot
>>> snap = MetricSnapshot(timestamp=0, duration=1.0, values={})
>>> e = SnapshotEvent(run_id="x", timestamp_ns=0, snapshot=snap)
>>> e.snapshot.duration
1.0

class rampa.events.ThresholdEvent class class class rampa.events.ThresholdEvent

Bases: EngineEvent

Threshold evaluation results.

>>> e = ThresholdEvent(run_id="x", timestamp_ns=0, results=[])
>>> len(e.results)
0

Configuration¶

class rampa.config.Config class class class rampa.config.Config

Bases: BaseModel

Top-level test configuration.

Either scenarios or shortcut options (vus, duration, etc.) may be provided, but not both.

>>> import datetime
>>> cfg = Config(
...     scenarios={"smoke": ScenarioConfig(
...         executor="constant-vus",
...         vus=1,
...         duration=datetime.timedelta(seconds=10),
...     )},
... )
>>> "smoke" in cfg.scenarios
True
>>> cfg.scenarios["smoke"].executor
'constant-vus'

class rampa.config.ScenarioConfig class class class rampa.config.ScenarioConfig

Bases: BaseModel

Configuration for a single test scenario.

Each scenario maps to an executor that controls how iterations are scheduled. Executor-specific fields are validated by the executor, not by this model.

>>> import datetime
>>> cfg = ScenarioConfig(
...     executor="constant-vus",
...     vus=10,
...     duration=datetime.timedelta(seconds=30),
... )
>>> cfg.executor
'constant-vus'
>>> cfg.vus
10

class rampa.config.Stage class class class rampa.config.Stage

Bases: BaseModel

A single ramp stage with a target VU or rate count and duration.

>>> import datetime
>>> Stage(
...     duration=datetime.timedelta(seconds=30), target=100,
... ).target
100
>>> Stage(
...     duration=datetime.timedelta(minutes=1), target=0,
... ).duration
datetime.timedelta(seconds=60)

Metrics¶

class rampa.metrics.MetricSnapshot class class class rampa.metrics.MetricSnapshot

Bases: object

Frozen snapshot of aggregated metric values at a point in time.

>>> snap = MetricSnapshot(
...     timestamp=0,
...     duration=10.0,
...     values={"reqs": {"count": 100.0, "rate": 10.0}},
... )
>>> snap.values["reqs"]["rate"]
10.0

class rampa.thresholds.ThresholdResult class class class rampa.thresholds.ThresholdResult

Bases: object

Result of evaluating one threshold.

>>> r = ThresholdResult(source="p(95)<500", passed=True, lhs=450.0, rhs=500.0)
>>> r.passed
True

class rampa.thresholds.ThresholdExpression class class class rampa.thresholds.ThresholdExpression

Bases: object

A parsed threshold expression.

>>> expr = parse_threshold("p(95)<500")
>>> expr.aggregation
'p(95)'
>>> expr.operator
'<'
>>> expr.value
500.0