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.

run_id: str
attribute
attribute
run_id: str

Identifier the registry keys this run by.

script_path: str
attribute
attribute
script_path: str

Path of the test script the run was started from.

started_at: float
attribute
attribute
started_at: float

Start time in monotonic seconds, for measuring elapsed time rather than wall clock.

runtime: RuntimeRun | None = None
attribute
attribute
runtime: RuntimeRun | None = None

Live async state while the run is in flight. None before the tasks are attached and after completion releases them.

result: RunResult | None = None
attribute
attribute
result: RunResult | None = None

Final result once the run finishes. None while it is still running.

events: list[EngineEvent]
attribute
attribute
events: list[EngineEvent]

Events drained from the run so far, in emission order.

Examples

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

Return True if the run has finished.

>>> r = RunRecord(run_id="x", script_path="t.py", started_at=0.0)
>>> r.is_complete
False
__init__(run_id, script_path, started_at, runtime=None, result=None, events=[])
method
method
__init__(run_id, script_path, started_at, runtime=None, result=None, events=[])
Parameters:
Return type:

None

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

Bases: object

Non-serializable async state for an active run.

Held only while the run is in flight; the registry drops it on completion so the metric engine threads and client sessions behind it are released.

controller: RunController
attribute
attribute
controller: RunController

Control surface for the run, used to stop it and read live snapshots.

wait_task: Task[RunResult]
attribute
attribute
wait_task: Task[RunResult]

Task awaiting the run’s final result.

event_task: Task[None]
attribute
attribute
event_task: Task[None]

Task draining the run’s event stream into the record’s events list.

Examples

>>> import rampa.mcp.registry
__init__(controller, wait_task, event_task)
method
method
__init__(controller, wait_task, event_task)
Parameters:
Return type:

None

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()
[]
__init__()
method
method
__init__()
Return type:

None

register(record)
method
method
register(record)

Add a new run to the registry.

Parameters:

  • record (RunRecord) – The run record to register.

  • RunRegistry() (>>> reg =)

  • RunRecord(run_id="r1" (>>> rec =)

  • script_path="t.py"

  • started_at=0.0)

  • reg.register(rec) (>>>)

  • rec (>>> reg.get("r1") is)

  • True

Return type:

None

get(run_id)
method
method
get(run_id)

Look up a run by ID.

Parameters:

run_id (str) – The run identifier.

Returns:

The run record, or None if not found.

Return type:

RunRecord | None

list_all()
method
method
list_all()

Return all registered runs.

Returns:

All run records.

Return type:

list[RunRecord]

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

The tools and resources hand back rampa’s own event types, which the library API Reference documents: a finished run arrives as a RunResult carrying a RunStatus, and a live run streams PhaseEvent, SnapshotEvent, and ThresholdEvent.

Configuration

You describe a run with the same Config the CLI loads, built from ScenarioConfig entries and their Stage ramps.

Metrics

Metric and threshold queries answer with a MetricSnapshot, a ThresholdResult per threshold, and the ThresholdExpression each one was parsed from.