API Reference#

This section provides auto-generated API documentation for every public module in the openswmm_mcp package.

Core Modules #

openswmm_mcp#

OpenSWMM MCP Server — SWMM engine tools for LLM-driven stormwater modeling.

openswmm_mcp.server#

The FastMCP composition root. Mounts every tool / resource / prompt sub-server.

Root MCP server module – creates the FastMCP instance and mounts sub-servers.

The mcp object defined here is the single entry point used by __main__.py (python -m openswmm_mcp) and by any ASGI / transport runner that needs a reference to the application.

Sub-servers are mounted with namespace prefixes so that tool names are scoped to their domain (e.g. lifecycle_open_model, query_get_node). Resources and prompts are mounted without a namespace so their URIs remain short.

openswmm_mcp.config#

Server configuration via environment variables (OPENSWMM_MCP prefix).

class openswmm_mcp.config.ServerSettings(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_prefix_target: EnvPrefixTarget | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_nested_max_split: int | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_show_env_vars: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None, _cli_ignore_unknown_args: bool | None = None, _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None, _cli_shortcuts: Mapping[str, str | list[str]] | None = None, _secrets_dir: PathType | None = None, _build_sources: tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None = None, *, working_dir: str = './', max_sessions: int = 5, transport: str = 'stdio', http_port: int = 8080, log_level: str = 'INFO', oauth_issuer: str | None = None, oauth_audience: str | None = None, jwt_jwks_url: str | None = None)[source]#

Bases: BaseSettings

OpenSWMM-MCP server settings.

Every field can be overridden with an environment variable prefixed by OPENSWMM_MCP_. For example, OPENSWMM_MCP_MAX_SESSIONS=10.

http_port: int#

jwt_jwks_url: str | None#

log_level: str#

max_sessions: int#

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_show_env_vars': False, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'OPENSWMM_MCP_', 'env_prefix_target': 'variable', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

oauth_audience: str | None#

oauth_issuer: str | None#

transport: str#

working_dir: str#

openswmm_mcp.session#

Session management for the OpenSWMM MCP server.

Provides SimSession (a thin wrapper around a Backend) and SessionManager (a concurrent-safe registry of active sessions).

class openswmm_mcp.session.SessionManager(max_sessions: int = 5, working_dir: str = './')[source]#

Bases: object

Thread/async-safe registry of SimSession instances.

Parameters:

max_sessions – Maximum number of concurrent sessions. Attempts to exceed this limit raise ToolError.
working_dir – Root directory under which per-session working directories are placed.

async cleanup_all() → None[source]#

Tear down every active session.

Intended to be called during server shutdown so that engine handles are not leaked.

async close_session(session_id: str) → None[source]#

Clean up and remove a session.

Raises:: ToolError – If no session with session_id exists.

async create_session(session_id: str, inp_path: str, rpt_path: str | None = None, out_path: str | None = None, engine: str = 'openswmm') → SimSession[source]#

Create a new SimSession and register it.

The backend (and its solver) is created and its file paths are set, but the caller is responsible for opening / initializing / starting the engine.

Parameters:

session_id – Unique identifier for the session.
inp_path – Path to the SWMM .inp input file.
rpt_path – Optional path for the report file. Defaults to <inp_stem>.rpt next to the input file.
out_path – Optional path for the binary output file. Defaults to <inp_stem>.out next to the input file.
engine – Which backend to use: "openswmm" (default, full-feature) or "legacy" (EPA SWMM 5.x — basic query / forcing / mass balance only).

Returns:

The newly created (but not yet opened) session.

Return type:

SimSession

Raises:

ToolError – If max_sessions has been reached, session_id already exists, or engine is unknown.

async get_session(session_id: str) → SimSession[source]#

Retrieve an existing session by its identifier.

Raises:: ToolError – If no session with session_id exists.

async list_sessions() → list[dict[str, Any]][source]#

Return metadata for every active session.

Returns:: Each dict contains id, state, engine, and working_dir.
Return type:: list[dict]

class openswmm_mcp.session.SessionMeta(session: SimSession)[source]#

Bases: object

Read-only snapshot of static model metadata shared across tool calls.

Populated lazily from the backend the first time a field is read. Topology (node / link / subcatchment ids and counts) is fixed once the model is parsed, so we cache it once per session and avoid the N C-ABI crossings that scalar get_id loops incur.

All fields use Python lists rather than NumPy arrays so the cache is JSON-serialisable for diagnostic and debugging purposes. Tool authors needing numeric arrays should call the corresponding *_bulk accessor directly — those return fresh NumPy arrays per call (no shared state).

Variables:

n_gages (n_nodes / n_links / n_subcatchments / n_pollutants /) – Element counts (cached on first access; remain valid until the session’s topology is mutated and invalidate_meta() is called).
gage_ids (node_ids / link_ids / subcatch_ids / pollutant_ids /) – Lists of element IDs in canonical (index) order. Each list is populated from the corresponding *.get_ids_bulk() accessor when available (added in Phase 3), with a transparent fallback to per-element get_id when running against an older binding (notably the legacy backend, which exposes only the scalar form).

property gage_ids: list[str]#

property link_ids: list[str]#

property n_gages: int#

property n_links: int#

property n_nodes: int#

property n_pollutants: int#

property n_subcatchments: int#

property node_ids: list[str]#

property pollutant_ids: list[str]#

property subcatch_ids: list[str]#

class openswmm_mcp.session.SimSession(backend: Backend | None = None, state: str = 'created', working_dir: Path = <factory>, inp_path: str = '', rpt_path: str = '', out_path: str = '', model_builder: Any = None, _output_reader: Any = None, _meta: Any = None)[source]#

Bases: object

Wraps a Backend instance with lazy domain-accessor delegation.

The backend supplies solver, nodes, links, etc. and tools call those attributes through this session object as if it were the backend itself. Setting session.<attr> for arbitrary attributes (e.g. model_builder, output_reader) is still supported for tools that cache state on the session.

Parameters:

backend (openswmm_mcp.backends.base.Backend | None) – The engine backend wrapping the underlying solver and domain objects. Optional: None while in the "building" state (no solver exists yet — the session uses model_builder instead).
state (str) – Human-readable lifecycle label. One of "created", "opened", "initialized", "running", "ended", "closed", "building".
working_dir (pathlib.Path) – Scratch directory used for temporary / output files.
out_path (inp_path / rpt_path /) – File paths the session was created with. Stored on the session so tools that need them (e.g. the output reader) don’t depend on engine-specific solver attributes.
model_builder (Any) – Optional new-engine ModelBuilder when the session was created programmatically rather than from an .inp file. Always None on the legacy backend.

backend: Backend | None = None#

cleanup() → None[source]#

Safely tear down the solver regardless of current state.

The method walks backward through the engine lifecycle so that the solver is left fully destroyed even if it was mid-run when cleanup was requested. Individual lifecycle calls are wrapped in try / except so that one failure does not prevent subsequent teardown steps.

property engine_kind: str#

inp_path: str = ''#

invalidate_meta() → None[source]#

Drop the cached metadata.

Should be called when model topology changes mid-session (e.g. after an editing.delete_object call). Most callers will never need this — topology is fixed for the lifetime of the simulation in normal usage.

property meta: SessionMeta#

Lazy cache of static (topology + pollutant) metadata.

Computed on first access from the backend’s bulk getters. Safe to call from any state where the backend is open; raises an AttributeError (via __getattr__) on a backend-less building session.

Returns the same SessionMeta instance on subsequent calls, so cached id-lists and counts cost nothing after the first read.

model_builder: Any = None#

out_path: str = ''#

property output_reader: Any#

rpt_path: str = ''#

state: str = 'created'#

working_dir: Path#

openswmm_mcp.dependencies#

Cross-tool dependency helpers (session lookups, state guards, engine acquisition).

Lifespan management and dependency injection for the OpenSWMM MCP server.

openswmm_mcp.dependencies.get_env_manager(ctx: fastmcp.Context)[source]#

Extract the gym C{EnvManager} from the FastMCP lifespan context.

@param ctx: The FastMCP L{Context} injected into a tool handler. @type ctx: L{Context} @return: The shared interactive-env manager. @rtype: L{EnvManager<openswmm_mcp.gym_support.envs.EnvManager>} @raise ToolError: If the env manager is not available in the context.

openswmm_mcp.dependencies.get_job_manager(ctx: fastmcp.Context)[source]#

Extract the gym C{JobManager} from the FastMCP lifespan context.

@param ctx: The FastMCP L{Context} injected into a tool handler. @type ctx: L{Context} @return: The shared background-optimization job manager. @rtype: L{JobManager<openswmm_mcp.gym_support.jobs.JobManager>} @raise ToolError: If the job manager is not available in the context.

openswmm_mcp.dependencies.get_session_manager(ctx: fastmcp.Context) → SessionManager[source]#

Extract the SessionManager from the FastMCP lifespan context.

Parameters:: ctx – The FastMCP Context injected into a tool handler.
Returns:: The shared session manager instance.
Return type:: SessionManager
Raises:: ToolError – If the session manager is not available in the context.

openswmm_mcp.dependencies.get_settings(ctx: fastmcp.Context) → ServerSettings[source]#

Extract ServerSettings from the FastMCP lifespan context.

Parameters:: ctx – The FastMCP Context injected into a tool handler.
Returns:: The server configuration loaded at startup.
Return type:: ServerSettings
Raises:: ToolError – If settings are not available in the context.

openswmm_mcp.dependencies.require_gymnasium(feature: str = 'This tool') → None[source]#

Assert that the optional openswmm.gymnasium package is importable.

Gym tool modules import openswmm_gymnasium lazily so the server starts cleanly without the gym extra; this guard converts the eventual ImportError into an actionable ToolError instead.

Parameters:: feature – Human-readable name of the feature being requested, used in the error message (e.g. "gym_run_episode").
Raises:: ToolError – With code DEPENDENCY_MISSING when openswmm_gymnasium cannot be imported.

openswmm_mcp.dependencies.require_new_engine(session, feature: str) → None[source]#

Assert that session uses the new openswmm engine backend.

Tools that depend on new-engine-only APIs (ModelBuilder, ModelEditor, Controls, Inflows, Infrastructure, Quality, Spatial, Tables, Statistics, OutputReader, GeoPackage) call this guard to fail fast on legacy sessions with a clear error code instead of an obscure AttributeError from the backend.

Parameters:

session – A session object exposing engine_kind.
feature – Human-readable name of the feature being requested, used in the error message (e.g. "ModelBuilder", "Spatial coordinates").

Raises:

ToolError – With code NOT_SUPPORTED when the session was created with engine='legacy'.

openswmm_mcp.dependencies.require_state(session, *valid_states: str) → None[source]#

Assert that session is in one of valid_states.

Parameters:

session – A session object that exposes a .state attribute.
*valid_states – One or more acceptable state strings (e.g. "running", "paused").

Raises:

ToolError – If session.state is not among the accepted states.

async openswmm_mcp.dependencies.server_lifespan(server) → AsyncIterator[dict]#

FastMCP lifespan handler: bootstrap shared resources and tear them down on shutdown.

Yields a context dict containing:: session_manager: SessionManager shared across all tool calls. settings: ServerSettings loaded from the environment.

openswmm_mcp.models#

Pydantic response models for all OpenSWMM-MCP tool responses.

class openswmm_mcp.models.BuildingResult(*, status: str, element_type: str, element_id: str, index: int, message: str)[source]#

Bases: BaseModel

Acknowledgement after creating or modifying a model element.

element_id: str#

element_type: str#

index: int#

message: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: str#

class openswmm_mcp.models.CapacitySummaryItem(*, link_id: str, max_filling: float, max_flow: float, max_velocity: float, time_above_threshold: float, vol_flow: float = 0.0)[source]#

Bases: BaseModel

Per-link capacity / flow statistics.

link_id: str#

max_filling: float#

max_flow: float#

max_velocity: float#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

time_above_threshold: float#

vol_flow: float#

class openswmm_mcp.models.ConduitGeometry(*, xsect: CrossSectionInfo | None = None, slope: float | None = None, offset_up: float = 0.0, offset_dn: float = 0.0, initial_flow: float = 0.0, max_flow: float = 0.0, loss_coeff_inlet: float = 0.0, loss_coeff_outlet: float = 0.0, loss_coeff_avg: float = 0.0, flap_gate: bool = False, seep_rate: float = 0.0, culvert_code: int = 0, barrels: int = 1)[source]#

Bases: BaseModel

Full geometry descriptor for a CONDUIT link.

barrels: int#

culvert_code: int#

flap_gate: bool#

initial_flow: float#

loss_coeff_avg: float#

loss_coeff_inlet: float#

loss_coeff_outlet: float#

max_flow: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

offset_dn: float#

offset_up: float#

seep_rate: float#

slope: float | None#

xsect: CrossSectionInfo | None#

class openswmm_mcp.models.ConversionResultModel(*, session_id: str, object_type: str, object_id: str, new_type: str, cleared_fields: list[str], warnings: list[str])[source]#

Bases: BaseModel

Result of an in-place type conversion.

cleared_fields: list[str]#: Type-specific fields that were cleared during conversion.

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

new_type: str#: Human-readable new type name.

object_id: str#

object_type: str#: node or link.

session_id: str#

warnings: list[str]#: Non-fatal topology warnings (e.g. “model has no outfall”).

class openswmm_mcp.models.CrossSectionInfo(*, shape: int, shape_name: str, geom1: float, geom2: float = 0.0, geom3: float = 0.0, geom4: float = 0.0, geom_labels: dict[str, float]=<factory>)[source]#

Bases: BaseModel

Cross-section shape and geometry parameters for a conduit or weir.

geom1: float#

geom2: float#

geom3: float#

geom4: float#

geom_labels: dict[str, float]#: Mapping of semantic parameter name to value, e.g. {"diameter": 1.2}.

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

shape: int#: Integer shape code (see XSectShape enum).

shape_name: str#: Human-readable shape name, e.g. CIRCULAR.

class openswmm_mcp.models.ElementSearchResult(*, element_type: str, element_id: str, index: int)[source]#

Bases: BaseModel

Single match returned by an element search.

element_id: str#

element_type: str#

index: int#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class openswmm_mcp.models.ExportResult(*, status: str, path: str, format: str, record_count: int)[source]#

Bases: BaseModel

Acknowledgement after an export operation.

format: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path: str#

record_count: int#

status: str#

class openswmm_mcp.models.FloodingSummaryItem(*, node_id: str, max_overflow_rate: float, total_flood_volume: float, time_flooded: float, max_depth: float)[source]#

Bases: BaseModel

Per-node flooding statistics.

max_depth: float#

max_overflow_rate: float#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_id: str#

time_flooded: float#

total_flood_volume: float#

class openswmm_mcp.models.ForcingResult(*, status: str, target_type: str, element_id: str, variable: str, value: float, mode: str, persist: bool)[source]#

Bases: BaseModel

Acknowledgement after applying a forcing override.

element_id: str#

mode: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

persist: bool#

status: str#

target_type: str#

value: float#

variable: str#

class openswmm_mcp.models.GageConfigResult(*, session_id: str, gage_id: str, updated_fields: dict[str, str | float | int])[source]#

Bases: BaseModel

Returned after configuring a rain gage.

gage_id: str#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

session_id: str#

updated_fields: dict[str, str | float | int]#: Mapping of field name to the new value that was applied.

class openswmm_mcp.models.GageInfo(*, gage_id: str, index: int, data_source: str, rain_type: str, rainfall: float | None = None)[source]#

Bases: BaseModel

Properties and current state of a rain gage.

data_source: str#

gage_id: str#

index: int#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rain_type: str#

rainfall: float | None#

class openswmm_mcp.models.HotStartResult(*, status: str, path: str, message: str)[source]#

Bases: BaseModel

Result of a hot-start save or load operation.

message: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path: str#

status: str#

class openswmm_mcp.models.ImpactEntryModel(*, obj_type: int, obj_type_name: str, obj_idx: int, field: str, cascaded: bool)[source]#

Bases: BaseModel

One object that is affected by a deletion (cascade or nullification).

cascaded: bool#: True if the object was deleted; False if only the reference was nullified.

field: str#: Name of the cross-reference field that was affected.

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

obj_idx: int#: Zero-based index of the affected object.

obj_type: int#

0=node, 1=link, 2=subcatchment, 3=gage, 4=table, 5=transect, 6=inlet_usage.

Type:: Integer code

obj_type_name: str#: Human-readable object type name.

class openswmm_mcp.models.ImpactReportModel(*, session_id: str, object_type: str, object_id: str, dry_run: bool, node_count: int, link_count: int, impacts: list[ImpactEntryModel])[source]#

Bases: BaseModel

Result of a deletion impact analysis or a deletion operation.

dry_run: bool#: True when the analysis was non-destructive (no objects were deleted).

impacts: list[ImpactEntryModel]#: Objects that were (or would be) affected.

link_count: int#: Number of links remaining after deletion (same as before for dry_run).

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_count: int#: Number of nodes remaining after deletion (same as before for dry_run).

object_id: str#: Identifier of the deleted / analysed object.

object_type: str#: Type of the deleted / analysed object (node, link, etc.).

session_id: str#

class openswmm_mcp.models.LinkFlowEntryModel(*, link_id: str, link_type: str, max_flow: float, max_velocity: float, max_filling: float, total_volume: float, surcharge_time: float)[source]#

Bases: BaseModel

One row of the Link Flow Summary table.

link_id: str#

link_type: str#

max_filling: float#

max_flow: float#

max_velocity: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

surcharge_time: float#

total_volume: float#

Bases: BaseModel

Properties and current state of a single link.

capacity: float | None#

conduit: ConduitGeometry | None#

depth: float | None#

flow: float | None#

from_node: str#

hydraulic_power: float | None#

index: int#

initial_flow: float | None#

length: float | None#

link_id: str#

link_type: str#

max_flow: float | None#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

offset_dn: float | None#

offset_up: float | None#

orifice: OrificeGeometry | None#

pump: PumpGeometry | None#

pump_cycles: int | None#

pump_on_time: float | None#

pump_volume: float | None#

roughness: float | None#

slope: float | None#

to_node: str#

velocity: float | None#

weir: WeirGeometry | None#

xsect: CrossSectionInfo | None#

class openswmm_mcp.models.MassBalanceResult(*, session_id: str, runoff_continuity_error: float, routing_continuity_error: float, quality_continuity_error: float | None = None, quality_continuity_errors: dict[str, float] | None = None, runoff_total: dict[str, float], routing_total: dict[str, float], routing_stats: dict[str, float] | None = None, max_courant: float | None = None, engine_kind: str | None = None, unsupported_fields: list[str] | None = None)[source]#

Bases: BaseModel

Continuity errors and volumetric totals.

engine_kind: str | None#

max_courant: float | None#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

quality_continuity_error: float | None#

quality_continuity_errors: dict[str, float] | None#

routing_continuity_error: float#

routing_stats: dict[str, float] | None#

routing_total: dict[str, float]#

runoff_continuity_error: float#

runoff_total: dict[str, float]#

session_id: str#

unsupported_fields: list[str] | None#

class openswmm_mcp.models.ModelSummary(*, session_id: str, state: str, engine: str = 'openswmm', node_count: int, link_count: int, subcatchment_count: int, gage_count: int, pollutant_count: int, flow_units: str, route_model: str, start_time: float, end_time: float, routing_step: float)[source]#

Bases: BaseModel

Returned after opening or inspecting a SWMM model.

end_time: float#

engine: str#

flow_units: str#

gage_count: int#

link_count: int#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_count: int#

pollutant_count: int#

route_model: str#

routing_step: float#

session_id: str#

start_time: float#

state: str#

subcatchment_count: int#

class openswmm_mcp.models.NodeFloodingEntryModel(*, node_id: str, node_type: str, max_depth: float, max_overflow_rate: float, total_flood_volume: float, time_flooded: float)[source]#

Bases: BaseModel

One row of the Node Flooding Summary table.

max_depth: float#

max_overflow_rate: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_id: str#

node_type: str#

time_flooded: float#

total_flood_volume: float#

Bases: BaseModel

Properties and current state of a single node.

crown_elev: float | None#

degree: int | None#

depth: float | None#

full_volume: float | None#

head: float | None#

index: int#

initial_depth: float | None#

invert_elev: float | None#

lateral_inflow: float | None#

losses: float | None#

max_depth: float | None#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_id: str#

node_type: str#

outfall: OutfallGeometry | None#

outfall_route_to: int | None#

outflow: float | None#

overflow: float | None#

ponded_area: float | None#

ponded_quality: dict[str, float] | None#

storage: StorageGeometry | None#

surcharge_depth: float | None#

volume: float | None#

class openswmm_mcp.models.OrificeGeometry(*, xsect: CrossSectionInfo | None = None, offset_up: float = 0.0, offset_dn: float = 0.0)[source]#

Bases: BaseModel

Geometry descriptor for an ORIFICE link.

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

offset_dn: float#

offset_up: float#

xsect: CrossSectionInfo | None#

class openswmm_mcp.models.OutfallGeometry(*, outfall_type: int = 0, outfall_type_name: str = 'FREE', param: float = 0.0, flap_gate: bool = False)[source]#

Bases: BaseModel

Boundary-condition geometry for an OUTFALL node.

flap_gate: bool#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

outfall_type: int#

outfall_type_name: str#: One of FREE, NORMAL, FIXED, TIDAL, TIMESERIES.

param: float#: Stage value for FIXED outfalls; 0 for others.

class openswmm_mcp.models.PollutantInfo(*, pollutant_id: str, index: int, units: int, units_name: str, kdecay: float, rain_conc: float, gw_conc: float, init_conc: float, rdii_conc: float, mwt: float, snow_only: bool, co_pollutant_idx: int, co_pollutant_frac: float)[source]#

Bases: BaseModel

Definition and properties of a single pollutant.

co_pollutant_frac: float#

co_pollutant_idx: int#

gw_conc: float#

index: int#

init_conc: float#

kdecay: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

mwt: float#

pollutant_id: str#

rain_conc: float#

rdii_conc: float#

snow_only: bool#

units: int#

0=MG/L, 1=UG/L, 2=#/L.

Type:: Concentration units code

units_name: str#

class openswmm_mcp.models.PropertyUpdateResult(*, session_id: str, element_type: str, element_id: str, updated_fields: dict[str, float | int | str])[source]#

Bases: BaseModel

Returned after updating element properties in-place.

element_id: str#

element_type: str#: node, link, or subcatchment.

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

session_id: str#

updated_fields: dict[str, float | int | str]#: Mapping of field name to the new value that was applied.

class openswmm_mcp.models.PumpEntryModel(*, link_id: str, pump_curve_idx: int, num_startups: int, total_on_time: float, total_volume: float, pct_time_on: float)[source]#

Bases: BaseModel

One row of the Pump Summary table.

link_id: str#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

num_startups: int#

pct_time_on: float#

pump_curve_idx: int#

total_on_time: float#

total_volume: float#

class openswmm_mcp.models.PumpGeometry(*, pump_curve_idx: int = -1, init_state_on: bool = False, offset_up: float = 0.0, offset_dn: float = 0.0)[source]#

Bases: BaseModel

Geometry descriptor for a PUMP link.

init_state_on: bool#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

offset_dn: float#

offset_up: float#

pump_curve_idx: int#

class openswmm_mcp.models.QualityContinuityModel(*, pollutant_id: str, continuity_error_pct: float, seep_loss: float, evap_loss: float)[source]#

Bases: BaseModel

Quality Routing Continuity entry for a single pollutant.

continuity_error_pct: float#

evap_loss: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pollutant_id: str#

seep_loss: float#

class openswmm_mcp.models.ReportSnapshotModel(*, routing_diagnostics: RoutingDiagnosticsModel, runoff_continuity: RunoffContinuityModel, routing_continuity: RoutingContinuityModel, quality_continuity: list[QualityContinuityModel], node_flooding: list[NodeFloodingEntryModel], storage_summary: list[NodeFloodingEntryModel], link_flow_summary: list[LinkFlowEntryModel], pump_summary: list[PumpEntryModel], subcatchment_summary: list[SubcatchmentEntryModel])[source]#

Bases: BaseModel

Full programmatic equivalent of the SWMM .rpt report file.

link_flow_summary: list[LinkFlowEntryModel]#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_flooding: list[NodeFloodingEntryModel]#

pump_summary: list[PumpEntryModel]#

quality_continuity: list[QualityContinuityModel]#

routing_continuity: RoutingContinuityModel#

routing_diagnostics: RoutingDiagnosticsModel#

runoff_continuity: RunoffContinuityModel#

storage_summary: list[NodeFloodingEntryModel]#

subcatchment_summary: list[SubcatchmentEntryModel]#

class openswmm_mcp.models.RoutingContinuityModel(*, continuity_error_pct: float, dry_weather_inflow: float, wet_weather_inflow: float, groundwater_inflow: float, rdii_inflow: float, external_inflow: float, total_flooding: float, total_outflow: float, evaporation_loss: float, seepage_loss: float, initial_storage: float, final_storage: float)[source]#

Bases: BaseModel

Flow Routing Continuity table.

continuity_error_pct: float#

dry_weather_inflow: float#

evaporation_loss: float#

external_inflow: float#

final_storage: float#

groundwater_inflow: float#

initial_storage: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rdii_inflow: float#

seepage_loss: float#

total_flooding: float#

total_outflow: float#

wet_weather_inflow: float#

class openswmm_mcp.models.RoutingDiagnosticsModel(*, avg_time_step: float, min_time_step: float, max_time_step: float, n_steps: int, pct_not_converged: float, n_steps_not_converged: int, avg_iterations: float, max_courant: float)[source]#

Bases: BaseModel

Routing Time Step Summary — convergence and time-step statistics.

avg_iterations: float#

avg_time_step: float#

max_courant: float#

max_time_step: float#

min_time_step: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n_steps: int#

n_steps_not_converged: int#

pct_not_converged: float#

class openswmm_mcp.models.RunoffContinuityModel(*, continuity_error_pct: float, total_rainfall: float, total_evaporation: float, total_infiltration: float, total_runoff: float, total_snow_removal: float, initial_storage: float, final_storage: float)[source]#

Bases: BaseModel

Runoff Quantity Continuity table.

continuity_error_pct: float#

final_storage: float#

initial_storage: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

total_evaporation: float#

total_infiltration: float#

total_rainfall: float#

total_runoff: float#

total_snow_removal: float#

class openswmm_mcp.models.SessionListItem(*, session_id: str, state: str, engine: str = 'openswmm', node_count: int, link_count: int, subcatchment_count: int)[source]#

Bases: BaseModel

Compact session descriptor used in list-sessions responses.

engine: str#

link_count: int#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_count: int#

session_id: str#

state: str#

subcatchment_count: int#

class openswmm_mcp.models.SimulationResult(*, session_id: str, elapsed_wall_time: float, steps_completed: int, runoff_continuity_error: float, routing_continuity_error: float, quality_continuity_error: float | None = None, engine_kind: str | None = None, unsupported_fields: list[str] | None = None)[source]#

Bases: BaseModel

Returned when a full simulation run completes.

elapsed_wall_time: float#

engine_kind: str | None#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

quality_continuity_error: float | None#

routing_continuity_error: float#

runoff_continuity_error: float#

session_id: str#

steps_completed: int#

unsupported_fields: list[str] | None#

class openswmm_mcp.models.SpatialResult(*, element_type: str, element_id: str, x: float | None = None, y: float | None = None, vertices: list[tuple[float, float]] | None = None)[source]#

Bases: BaseModel

Coordinates and vertex geometry for a model element.

element_id: str#

element_type: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

vertices: list[tuple[float, float]] | None#

x: float | None#

y: float | None#

class openswmm_mcp.models.StepResult(*, session_id: str, elapsed: float, current_time: float, completed: bool, steps_taken: int)[source]#

Bases: BaseModel

Returned after executing one or more simulation steps.

completed: bool#

current_time: float#

elapsed: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

session_id: str#

steps_taken: int#

class openswmm_mcp.models.StorageGeometry(*, storage_type: str = 'functional', curve_idx: int | None = None, functional_a: float | None = None, functional_b: float | None = None, functional_c: float | None = None, seep_rate: float = 0.0, exfil_suction: float | None = None, exfil_ksat: float | None = None, exfil_imd: float | None = None)[source]#

Bases: BaseModel

Storage-unit geometry for a STORAGE node.

curve_idx: int | None#

exfil_imd: float | None#

exfil_ksat: float | None#

exfil_suction: float | None#

functional_a: float | None#: Coefficient A in Area = A * Depth^B + C.

functional_b: float | None#

functional_c: float | None#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

seep_rate: float#

storage_type: str#: curve if a stage-area curve is used, functional otherwise.

class openswmm_mcp.models.SubcatchmentEntryModel(*, subcatch_id: str, total_precip: float, total_runoff_vol: float, max_runoff_rate: float, runoff_coefficient: float)[source]#

Bases: BaseModel

One row of the Subcatchment Runoff Summary table.

max_runoff_rate: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runoff_coefficient: float#

subcatch_id: str#

total_precip: float#

total_runoff_vol: float#

Bases: BaseModel

Properties and current state of a single subcatchment.

area: float | None#

depth: float | None#

imperv_pct: float | None#

index: int#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rainfall: float | None#

runoff: float | None#

slope: float | None#

subcatch_id: str#

width: float | None#

class openswmm_mcp.models.SystemSummary(*, session_id: str, state: str, engine: str = 'openswmm', node_count: int, link_count: int, subcatchment_count: int, gage_count: int, pollutant_count: int, flow_units: str, route_model: str, start_time: float, end_time: float, routing_step: float, current_time: float | None = None, surcharge_method: str | None = None, dps_celerity: float | None = None, dps_alpha: float | None = None, dps_decay_time: float | None = None, event_count: int | None = None, steady_state_skip: bool | None = None)[source]#

Bases: BaseModel

Full system-level summary including optional runtime state.

current_time: float | None#

dps_alpha: float | None#

dps_celerity: float | None#

dps_decay_time: float | None#

end_time: float#

engine: str#

event_count: int | None#

flow_units: str#

gage_count: int#

link_count: int#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_count: int#

pollutant_count: int#

route_model: str#

routing_step: float#

session_id: str#

start_time: float#

state: str#

steady_state_skip: bool | None#

subcatchment_count: int#

surcharge_method: str | None#

class openswmm_mcp.models.TimeSeries(*, element_type: str, element_id: str, variable: str, timestamps: list[float], values: list[float], units: str)[source]#

Bases: BaseModel

Variable time-series for a single element.

element_id: str#

element_type: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

timestamps: list[float]#

units: str#

values: list[float]#

variable: str#

class openswmm_mcp.models.WeirGeometry(*, xsect: CrossSectionInfo | None = None, crest_height: float = 0.0, discharge_coeff: float = 0.0, end_contractions: float = 0.0, offset_up: float = 0.0, offset_dn: float = 0.0)[source]#

Bases: BaseModel

Geometry descriptor for a WEIR link.

crest_height: float#

discharge_coeff: float#

end_contractions: float#

model_config = {'from_attributes': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

offset_dn: float#

offset_up: float#

xsect: CrossSectionInfo | None#

openswmm_mcp.errors#

Standardised error codes and helper utilities.

class openswmm_mcp.errors.ErrorCode[source]#

Bases: object

Canonical error-code strings returned in tool responses.

BAD_PARAM: str = 'BAD_PARAM'#

DEPENDENCY_MISSING: str = 'DEPENDENCY_MISSING'#

ELEMENT_NOT_FOUND: str = 'ELEMENT_NOT_FOUND'#

ENGINE_ERROR: str = 'ENGINE_ERROR'#

INVALID_STATE: str = 'INVALID_STATE'#

MAX_SESSIONS_REACHED: str = 'MAX_SESSIONS_REACHED'#

NOT_FOUND: str = 'NOT_FOUND'#

NOT_SUPPORTED: str = 'NOT_SUPPORTED'#

SESSION_NOT_FOUND: str = 'SESSION_NOT_FOUND'#

STALE_OBJECT: str = 'STALE_OBJECT'#

VALIDATION_ERROR: str = 'VALIDATION_ERROR'#

openswmm_mcp.errors.engine_error_response(exc: BaseException) → dict[str, Any][source]#

Convert an arbitrary exception into a standardised error dict.

Returns a dict suitable for returning directly from a tool handler:

{
    "error": "ENGINE_ERROR",
    "message": "<str(exc)>",
    "detail": "<traceback lines>",
}

openswmm_mcp.errors.raise_stale_object_as_tool_error(exc: BaseException) → None[source]#

Raise a translated ToolError when exc is a StaleObjectError.

Helper for tool authors who want a single-line re-raise inside a narrow try/except StaleObjectError (or a broad except Exception) clause. Returns normally — i.e. is a no-op — when the exception is not a stale-object error, so the caller can raise the original after this returns.

async openswmm_mcp.errors.resolve_index(accessor: Any, element_id: Any, kind: str = 'Element') → int[source]#

Resolve a string element id to its engine index off the event loop.

The handle-based engine’s get_index raises ElementNotFoundError (a KeyError subclass) when an id is unknown; this translates that into a clean ELEMENT_NOT_FOUND ToolError. Integer ids pass through unchanged (callers may already hold an index).

Parameters:

accessor – An engine collection exposing get_index(id) (e.g. session.nodes / session.links).
element_id – The id to resolve (str) or an existing index (int).
kind – Human-readable element kind for the error message.

openswmm_mcp.errors.translate_stale_object(exc: BaseException) → ToolError | None[source]#

Return a ToolError if exc is an engine StaleObjectError, else None.

Lets tool code do:

try:
    value = await asyncio.to_thread(lambda: session.nodes[idx].depth)
except Exception as e:
    translated = translate_stale_object(e)
    if translated is not None:
        raise translated
    raise

Or use raise_stale_object_as_tool_error() as a one-liner re-raise inside a focused try/except.

openswmm_mcp.auth#

OAuth / JWT authentication for the OpenSWMM MCP HTTP transport.

Authentication strategy#

stdio transport – No authentication is applied. The server runs as a local subprocess and inherits the caller’s OS-level identity, so adding token verification would only add friction with no security benefit.
http / sse transport – Three schemes are supported, selected by the ServerSettings fields jwt_jwks_url and oauth_issuer:
1. JWT only (jwt_jwks_url is set): incoming requests must carry a Bearer token whose signature is verified against the JSON Web Key Set at the given URL.
2. OAuth only (oauth_issuer is set): a full OAuth 2.0 authorization code / client-credentials flow is used, with the issuer URL serving as the OpenID Connect discovery root.
3. Both (both fields set): the two providers are combined so that a request succeeds if either verifier accepts the credential.
4. Neither set: the server starts without authentication. This is useful for local development or environments that handle auth at an upstream reverse-proxy layer.

The fastmcp[auth] extras package must be installed for any of the authenticated modes. If the extras are missing and auth is requested, the function raises an ImportError with an actionable message.

openswmm_mcp.auth.create_auth(settings: ServerSettings) → Any | None[source]#

Build an authentication provider based on settings, or return None.

Parameters:

settings – The ServerSettings loaded at startup.

Returns:

An auth provider instance understood by fastmcp.FastMCP, or
None if no authentication should be applied.

Raises:

ImportError – If authentication is requested but the fastmcp[auth] extras are not installed.

Backend Modules #

The backend layer abstracts the underlying SWMM engine so that the same tool surface can drive either the refactored openswmm.engine (v6.0) or the legacy SWMM 5 solver.

openswmm_mcp.backends.base#

Backend protocol definition.

Defines the interface every engine backend exposes to tools. Both backends present the v1 Pythonic surface of openswmm.engine:

Container protocol on every collection: len(backend.nodes), backend.nodes[id_or_idx], for n in backend.nodes, id_or_idx in backend.nodes.
Item access returns wrapper objects exposing property-style attribute reads/writes: backend.nodes["J1"].depth, backend.nodes["J1"].lateral_inflow = 0.5.
Sub-views per type: node.stats, node.storage, node.outfall, link.pump, link.weir, subcatchment.infiltration, etc. Available only on the openswmm backend; the legacy backend raises AttributeError on these.

The legacy adapter has scalar accessor methods (get_depth(idx), get_type(idx), …) on each collection — these are not a public transition shim, they are the SWMM-5-toolkit dispatch path that the _LegacyNode / _LegacyLink / etc. wrappers call into. Treat them as private implementation; the public surface is the v1 shape described above.

class openswmm_mcp.backends.base.Backend(*args, **kwargs)[source]#

Bases: Protocol

Engine-agnostic surface that tools call against.

New-engine-only attributes (editor, statistics, spatial, tables, patterns, controls, inflows, infrastructure, quality, save_schedule) raise AttributeError on the legacy backend; tools that need them must guard with openswmm_mcp.dependencies.require_new_engine().

Variables:

engine_kind (str) – "openswmm" or "legacy". Tools that require the new engine check this via openswmm_mcp.dependencies.require_new_engine().
solver – Lifecycle-managing solver handle. Has open / initialize / start / step / end / report / close / destroy, plus elapsed, state, start_datetime, end_datetime, current_datetime, and (openswmm only) steps() / stride(n) / until(target).
pollutants (nodes / links / subcatchments / gages /) – v1-shape collections: len(...), [key] returning wrappers with property-style attribute access, iteration, and in.
mass_balance – Continuity-error queries returning fractions (0.001 = 0.1 %), accessible via property-style mass_balance.runoff_continuity_error.
forcing – Runtime forcing dispatcher. On legacy a subset of methods is available; unsupported methods raise NotImplementedError which the forcing tool translates to a clear NOT_SUPPORTED ToolError.
hotstart – save(solver, path) and open(path) returning an object with apply(solver). Legacy backend wraps solver.save_hotstart / solver.use_hotstart to match this shape.

engine_kind: str#

property forcing: Any#

property gages: Any#

property hotstart: Any#

property links: Any#

property mass_balance: Any#

property nodes: Any#

property pollutants: Any#

property solver: Any#

property subcatchments: Any#

openswmm_mcp.backends.openswmm#

OpenSWMM (new engine) backend.

The v1 Solver already exposes every domain collection as a typed property (solver.nodes, solver.links, solver.subcatchments, solver.gages, solver.pollutants, solver.tables, solver.patterns, solver.inflows, solver.controls, solver.forcing, solver.infrastructure, solver.spatial, solver.quality, solver.statistics, solver.mass_balance, solver.editor, solver.save_schedule).

This backend is therefore a thin pass-through: every attribute that is not part of the backend’s own state (_solver, engine_kind) delegates straight to the underlying Solver.

Tools call session.<domain> (which resolves through __getattr__ to this backend’s __getattr__, which in turn forwards to solver.<domain>).

class openswmm_mcp.backends.openswmm.OpenSwmmBackend(inp_path: str, rpt_path: str, out_path: str)[source]#

Bases: object

Backend that delegates straight to the new openswmm.engine Solver.

engine_kind = 'openswmm'#

classmethod from_solver(solver: openswmm.engine.Solver) → OpenSwmmBackend[source]#: Wrap an already-constructed Solver (e.g. from ModelBuilder.to_solver).

property hotstart: _OpenSwmmHotstart#

v1-shape hot-start wrapper (.save(solver, path) / .open(path) -> HotStart).

Bypasses the __getattr__ pass-through because the v1 Solver has no hotstart attribute of its own — this is the wrapper defined locally to keep the tool layer uniform across backends.

property solver: openswmm.engine.Solver#

openswmm_mcp.backends.legacy#

Legacy SWMM (EPA SWMM 5.x) backend.

Translates the indexed-getter / domain-collection shape that tools call against onto the legacy property-based LegacyNode.depth API exposed by openswmm.legacy.engine.

Translations that need to happen:

Lifecycle: legacy Solver.initialize() does open + start in one call, whereas the new engine treats them separately. This adapter routes open / initialize / start so that callers get the same three-phase shape regardless of which engine is underneath.
Stepping: legacy step() returns (elapsed_days, current_datetime); this adapter caches those and returns bool (continue-or-not) to match the new engine.
Mass balance: legacy reports continuity errors in percent; this adapter divides by 100 so callers always see the new-engine fraction (0.001 = 0.1 %).
Hotstart: legacy Solver.use_hotstart / save_hotstart are exposed via a class that mirrors the new-engine HotStart.save / open / apply three-method shape so the hotstart tools work uniformly.

Anything genuinely missing from legacy (ModelBuilder, ModelEditor, Controls, Inflows, Infrastructure, Quality, Spatial, Tables, Statistics, OutputReader, GeoPackage) is not implemented here. The session’s __getattr__ falls through to the backend, AttributeError bubbles up, and the tool’s require_new_engine(...) guard raises a clean NOT_SUPPORTED error before the AttributeError path is ever hit.

class openswmm_mcp.backends.legacy.LegacyBackend(inp_path: str, rpt_path: str, out_path: str)[source]#

Bases: object

Backend that wraps openswmm.legacy.engine.

engine_kind = 'legacy'#

property solver: _LegacySolverAdapter#

Utility Modules #

openswmm_mcp._util.formatting#

Formatting helpers for converting engine data into human-/LLM-friendly strings.

openswmm_mcp._util.formatting.format_elapsed(days: float) → str[source]#

Convert a decimal number of days to an "HH:MM:SS" string.

Parameters:: days – Elapsed time expressed as fractional days (e.g. 0.5 == 12 hours).
Returns:: A string in "HH:MM:SS" format.
Return type:: str

Examples

>>> format_elapsed(0.5)
'12:00:00'
>>> format_elapsed(1.25)
'30:00:00'

openswmm_mcp._util.formatting.format_julian(julian: float) → str[source]#

Convert a Julian Day Number to an ISO-style datetime string.

Parameters:: julian – A Julian Day Number (e.g. 2_460_000.5).
Returns:: Datetime formatted as "YYYY-MM-DD HH:MM:SS".
Return type:: str

Examples

>>> format_julian(2451545.0)
'2000-01-01 12:00:00'

openswmm_mcp._util.formatting.ndarray_to_list(arr: Any) → list[source]#

Convert a NumPy ndarray (or None) to a plain Python list.

Parameters:: arr – A NumPy array, a sequence, or None.
Returns:: A plain Python list. Returns an empty list when arr is None.
Return type:: list

openswmm_mcp._util.formatting.paginate_list(items: list, start_index: int = 0, limit: int | None = None) → tuple[list, dict[str, int | bool]][source]#

Slice items using start_index/limit for tool responses.

This is the canonical pagination primitive for the MCP server’s O(n_elements) list-returning tools (get_node_info, get_link_info, find_elements, output_*_results …). Tools call paginate_list after the bulk fetch — slicing is cheap relative to the fetch and ensures every caller observes consistent pagination semantics.

Parameters:

items – The source list (already materialised by the tool — pagination does not lazy-fetch).
start_index – Zero-based offset of the first item to return. Negative values are clamped to 0; values past the end produce an empty slice.
limit – Maximum number of items in the returned slice, or None for “no limit” (the whole tail from start_index). Non-positive values produce an empty slice (cleaner than raising — callers often build URLs from user input where 0 means “do not return”).

Returns:

(slice, meta) where meta carries:

total — original item count

start_index — clamped offset actually used

limit — limit applied (-1 for “no limit”)

returned — len(slice)

has_more — True if items remain after the slice

Return type:

tuple[list, dict]

Examples

>>> paginate_list([1, 2, 3, 4, 5], start_index=1, limit=2)
([2, 3], {'total': 5, 'start_index': 1, 'limit': 2, 'returned': 2, 'has_more': True})
>>> paginate_list([1, 2, 3], start_index=0, limit=None)[0]
[1, 2, 3]
>>> paginate_list([], start_index=0, limit=10)[0]
[]

openswmm_mcp._util.formatting.truncate_list(items: list, max_items: int = 1000) → tuple[list, bool][source]#

Return at most max_items elements from items.

Parameters:

items – The source list.
max_items – Maximum number of elements to keep. Defaults to 1000.

Returns:

A two-element tuple (truncated_list, was_truncated) where was_truncated is True when elements were dropped.

Return type:

tuple[list, bool]

Examples

>>> truncate_list([1, 2, 3], max_items=2)
([1, 2], True)
>>> truncate_list([1, 2, 3], max_items=5)
([1, 2, 3], False)

openswmm_mcp._util.validation#

Input validation utilities for MCP tool handlers.

openswmm_mcp._util.validation.resolve_path(path: str, working_dir: str) → Path[source]#

Resolve path against working_dir, expanding ~.

Parameters:

path – A file-system path that may be relative or contain ~.
working_dir – The directory used as the base when path is relative.

Returns:

A fully resolved, absolute Path.

Return type:

Path

Examples

>>> resolve_path("model.inp", "/data/projects")
PosixPath('/data/projects/model.inp')
>>> resolve_path("~/models/test.inp", "/ignored")
PosixPath('/home/user/models/test.inp')

openswmm_mcp._util.validation.validate_element_type(element_type: str) → str[source]#

Validate that element_type is a recognised SWMM element category.

The comparison is case-insensitive; the returned value is always lowercase.

Parameters:: element_type – A string such as "node", "Link", or "SUBCATCHMENT".
Returns:: The normalised (lowercase) element type.
Return type:: str
Raises:: ToolError – If element_type is not one of the accepted values.

openswmm_mcp._util.validation.validate_session_state(session, *valid_states: str, action: str = 'perform this action') → None[source]#

Raise ToolError if the session is not in an accepted state.

Parameters:

session – A session object exposing a .state attribute.
*valid_states – One or more acceptable state strings (e.g. "running", "paused").
action – A human-readable description of what the caller is trying to do, used in the error message. Defaults to "perform this action".

Raises:

ToolError – If session.state is not among valid_states.

Tool Modules #

Each tool module exposes a FastMCP sub-server whose tools are mounted under the matching namespace prefix (e.g. tools.nodes → nodes_* tools).

openswmm_mcp.tools.lifecycle#

Lifecycle tools for the OpenSWMM MCP server.

Provides tools for opening, running, stepping, and closing SWMM models.

async openswmm_mcp.tools.lifecycle.close_model(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Close and clean up a simulation session.

Tears down the solver (ending the run if necessary), releases all engine resources, and removes the session from the registry.

async openswmm_mcp.tools.lifecycle.events_add(ctx: fastmcp.Context, session_id: str = 'default', start_oadate: float = 0.0, end_oadate: float = 0.0) → dict#: Append a new event window (OADate decimal days).

async openswmm_mcp.tools.lifecycle.events_clear(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Remove every event window. Safe on an already-empty list.

async openswmm_mcp.tools.lifecycle.events_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of [EVENTS] rows in the model.

async openswmm_mcp.tools.lifecycle.events_get(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the start/end OADate of the I{index}-th event.

async openswmm_mcp.tools.lifecycle.events_remove(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Remove the I{index}-th event; trailing entries shift down.

async openswmm_mcp.tools.lifecycle.events_set(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0, start_oadate: float = 0.0, end_oadate: float = 0.0) → dict#: Overwrite the I{index}-th event window.

async openswmm_mcp.tools.lifecycle.get_open_diagnostics(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return validation errors and warnings recorded during a (lenient) open.

After open_model(..., lenient_open=True) the engine records post-parse validation problems instead of raising, leaving the session in the editable opened state. This tool reads those accumulators (Solver.open_errors / Solver.open_warnings) so callers can inspect and fix issues before initialising or editing the model. A strict open that succeeds leaves both lists empty. New engine only.

async openswmm_mcp.tools.lifecycle.get_simulation_state(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the current session and solver state.

Includes the session lifecycle state, the raw engine state code, and basic model metadata (counts and file path).

async openswmm_mcp.tools.lifecycle.get_simulation_time(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the current simulation timing information.

Includes start time, end time, current time, elapsed fraction, and the routing timestep (seconds).

async openswmm_mcp.tools.lifecycle.get_steady_state_skip(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return whether SKIP_STEADY_STATE routing skip is enabled.

async openswmm_mcp.tools.lifecycle.is_between_events(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return whether the current sim time falls inside a defined event window.

async openswmm_mcp.tools.lifecycle.list_sessions(ctx: fastmcp.Context) → list[dict]#

List all active simulation sessions.

Returns metadata for every session currently managed by the server.

async openswmm_mcp.tools.lifecycle.load_runoff_interface(ctx: fastmcp.Context, session_id: str = 'default', path: str = '') → dict#

Open the runoff interface file for reading (USE mode).

The file’s header is verified against the current model (subcatchment count, pollutant count, flow units).

Note

USE-mode auto-skip — making the engine bypass its own runoff computation when the file is open — is a follow-up to Phase 1b. Today’s USE mode is an advanced manual feature. After opening, the caller must drive the simulation and invoke read_runoff_step between step_simulation() calls (currently only exposed through the Python binding, not MCP). Most LLM workflows should prefer SAVE mode plus a downstream routing-only run that consumes the file via an external script.

Parameters:

session_id – Identifier of the session.
path – Path to an existing runoff interface file produced by a previous SAVE-mode run.

Returns:

{"status": "ok", "session_id": ..., "path": ..., "mode": "use", "warning": "..."} on success. The warning field documents the USE-mode caveat so an LLM caller is aware of the manual orchestration requirement.

Return type:

dict

Raises:

ToolError – Backend is the legacy engine, the path is empty, the file is missing, or its header does not match the current model.

async openswmm_mcp.tools.lifecycle.open_model(ctx: fastmcp.Context, inp_path: str, session_id: str = 'default', rpt_path: str | None = None, out_path: str | None = None, engine: str = 'openswmm', lenient_open: bool = False) → ModelSummary#

Open a SWMM model file and initialise the engine.

Creates a new simulation session, parses the .inp file, and prepares the engine for simulation. Returns a summary of the loaded model.

Parameters:

inp_path – Path to the SWMM .inp input file.
session_id – Identifier for the new session. Defaults to "default".
out_path (rpt_path /) – Optional report and binary output file paths. When omitted they default to <inp_stem>.rpt / <inp_stem>.out.
engine – Which engine to use. "openswmm" (default) is the modern engine with full feature support. "legacy" selects the EPA SWMM 5.x bindings shipped alongside it; only basic query / forcing / mass-balance / hot-start tools are supported on legacy sessions — advanced tools (model building, in-place editing, controls, infrastructure, quality, spatial, geopackage, output reader) return a NOT_SUPPORTED error.
lenient_open – When True (new engine only), perform a permissive open that records post-parse validation problems instead of raising, and leave the session in the editable opened state (the model is not initialised). Read the recorded issues with get_open_diagnostics() and fix them via the editing tools before running. Defaults to False (strict open followed by initialise, landing in initialized).

async openswmm_mcp.tools.lifecycle.run_for_steps(ctx: fastmcp.Context, session_id: str = 'default', max_steps: int = 100, progress_interval: int = 0) → StepResult#

Run up to max_steps steps using the v1 Solver.steps() iterator.

Slightly different from stride(max_steps): stride is one C call, while run_for_steps issues the steps inside a Python loop so progress can be reported (via ctx.report_progress) every progress_interval steps. Use stride for raw speed, this one when you want intermediate progress.

The simulation stops at whichever happens first: max_steps steps completed, or the engine reaches the end of the simulation. Auto-starts the solver if needed.

Parameters:

max_steps – Upper bound on steps to advance.
progress_interval – Emit progress every N steps (0 = no progress).

async openswmm_mcp.tools.lifecycle.run_simulation(ctx: fastmcp.Context, session_id: str = 'default') → SimulationResult#

Run the full simulation to completion.

Starts the solver (if not already started), steps through every timestep, and reports progress as a percentage. Returns continuity errors and timing.

async openswmm_mcp.tools.lifecycle.save_runoff_interface(ctx: fastmcp.Context, session_id: str = 'default', path: str = '') → dict#

Open the runoff interface file for writing (SAVE mode).

Call this before run_simulation() (or before the first step_simulation()). The engine auto-emits one record per runoff substep until the session is closed, at which point the file is finalised automatically — there is no separate “close” tool needed for ordinary flows.

Parameters:

session_id – Identifier of the session. Defaults to "default".
path – Output file path. Existing content is truncated. An empty string is rejected.

Returns:

{"status": "ok", "session_id": ..., "path": ..., "mode": "save"} on success.

Return type:

dict

Raises:

ToolError – Backend is the legacy engine (this feature requires the new engine), the path is empty, or the file could not be opened.

async openswmm_mcp.tools.lifecycle.set_steady_state_skip(ctx: fastmcp.Context, session_id: str = 'default', enabled: bool = False) → dict#

Enable or disable SKIP_STEADY_STATE routing.

When enabled the engine skips routing during periods with unchanged flows; useful for long dry-weather periods between rainfall events.

async openswmm_mcp.tools.lifecycle.step_simulation(ctx: fastmcp.Context, session_id: str = 'default', num_steps: int = 1) → StepResult#

Advance the simulation by one or more timesteps.

Automatically starts the solver if the session is in the ‘initialized’ state. Returns the current simulation time and whether the run completed.

async openswmm_mcp.tools.lifecycle.stride(ctx: fastmcp.Context, session_id: str = 'default', num_steps: int = 1) → StepResult#

Advance the simulation by N timesteps in a single engine call.

Faster than calling step_simulation() N times — the C engine loops internally, so we pay one asyncio.to_thread crossing regardless of N.

Auto-starts the solver if the session is in the initialized state.

Parameters:

session_id – Identifier of the simulation session.
num_steps – Number of timesteps to advance. Negative or zero raises a validation error.

async openswmm_mcp.tools.lifecycle.until_datetime(ctx: fastmcp.Context, session_id: str = 'default', target_iso: str = '') → StepResult#

Advance the simulation until the wall-clock simulation datetime reaches target_iso.

Maps to Solver.until(datetime). The engine stops at the next routing-step boundary >= the target datetime. Auto-starts the solver if needed.

Parameters:: target_iso – ISO-8601 datetime string (e.g. "2026-01-01T12:00:00"). Must be > the simulation start and <= the simulation end.

async openswmm_mcp.tools.lifecycle.until_elapsed(ctx: fastmcp.Context, session_id: str = 'default', seconds: float = 0.0) → StepResult#

Advance the simulation until at least seconds of sim-time have elapsed.

Maps to Solver.until(timedelta(seconds=seconds)). The engine stops at the next routing-step boundary >= the target. Auto-starts the solver if needed.

Parameters:: seconds – Target elapsed simulation time in seconds, measured from the start of the simulation (not from the current step).

openswmm_mcp.tools.query#

Query tools for the OpenSWMM MCP server.

Provides read-only tools for inspecting model elements (nodes, links, subcatchments, gages) and searching across the model.

async openswmm_mcp.tools.query.find_elements(ctx: fastmcp.Context, session_id: str = 'default', pattern: str | None = None, element_type: str | None = None) → list[ElementSearchResult]#

Search for model elements by ID pattern and/or type.

pattern is a Python regex matched against element IDs (case-insensitive). element_type restricts the search to "node", "link", "subcatchment", or "gage". Both are optional; when neither is given all elements are returned.

async openswmm_mcp.tools.query.get_gage_info(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str | None = None) → GageInfo | list[GageInfo]#

Return properties and state for one or all rain gages.

When gage_id is given, returns a single GageInfo. When omitted, returns a list of GageInfo for every rain gage in the model.

async openswmm_mcp.tools.query.get_link_info(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | None = None, start_index: int = 0, limit: int | None = None) → LinkInfo | list[LinkInfo]#

Return properties and state for one or all links.

When link_id is given, returns a single LinkInfo. When omitted, returns a list of LinkInfo for every link in the model.

Phase 4d adds start_index / limit for paginated reads of the all-mode response, mirroring get_node_info().

Parameters:

start_index – Zero-based offset of the first link to include.
limit – Maximum number of links returned, or None for “no limit”.

async openswmm_mcp.tools.query.get_node_info(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | None = None, properties: list[str] | None = None, start_index: int = 0, limit: int | None = None) → NodeInfo | list[NodeInfo]#

Return properties and state for one or all nodes.

When node_id is given, returns a single NodeInfo. When omitted, returns a list of NodeInfo for every node in the model. The optional properties list filters which fields are included (not yet implemented; reserved for future optimisation).

Phase 4d adds start_index and limit for paginated reads of the all-mode response. Both default to “no pagination” (return every node). Pagination is applied after the bulk fetch — the underlying engine still does one pass over the entire network regardless of slice — so callers can safely make many small paged calls without re-paying the bulk-fetch cost beyond the per-call Python-side slice.

Parameters:

start_index – Zero-based offset of the first node to include in the response. Negative values are clamped to 0.
limit – Maximum number of nodes returned. None (the default) means “no limit”; non-positive values produce an empty list.

async openswmm_mcp.tools.query.get_pollutant_info(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | None = None) → PollutantInfo | list[PollutantInfo]#

Return definition and properties for one or all pollutants.

When pollutant_id is given, returns a single PollutantInfo. When omitted, returns a list of PollutantInfo for every pollutant in the model. Valid in any non-closed session state.

async openswmm_mcp.tools.query.get_subcatchment_info(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | None = None) → SubcatchmentInfo | list[SubcatchmentInfo]#

Return properties and state for one or all subcatchments.

When subcatch_id is given, returns a single SubcatchmentInfo. When omitted, returns a list of SubcatchmentInfo for every subcatchment.

async openswmm_mcp.tools.query.get_system_summary(ctx: fastmcp.Context, session_id: str = 'default') → SystemSummary#

Return a full system summary including counts, options, and timing.

Provides an overview of the loaded model’s configuration and, if a simulation is running or has ended, the current simulation time.

openswmm_mcp.tools.forcing#

Forcing and control tools for the OpenSWMM MCP server.

Provides tools for applying runtime forcing overrides (rainfall, inflows, boundary conditions, etc.) and manipulating control rules / link settings during a running simulation.

async openswmm_mcp.tools.forcing.add_control_rule(ctx: fastmcp.Context, session_id: str = 'default', rule_text: str = '') → dict#

Add a new control rule to the running simulation.

The rule is specified in SWMM rule syntax and takes effect immediately.

Parameters:

session_id – Target simulation session.
rule_text – The control rule in SWMM rule syntax (e.g. "RULE R1\nIF NODE J1 DEPTH > 5\nTHEN PUMP P1 STATUS = ON").

async openswmm_mcp.tools.forcing.clear_forcing(ctx: fastmcp.Context, session_id: str = 'default', target_type: str | None = None, element_id: str | None = None) → dict#

Clear forcing overrides.

If both target_type and element_id are None (the default), all forcing overrides across the entire model are removed. Otherwise, only the forcing on the specified element is cleared.

Parameters:

session_id – Target simulation session.
target_type – Element category ("node", "link", "subcatchment", "gage"). Required when clearing a single element.
element_id – The name or index of the element whose forcing should be removed. Required when clearing a single element.

async openswmm_mcp.tools.forcing.get_climate_evap_rate(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the current climate-derived evaporation rate (read-only).

Reports the broadcast potential-evapotranspiration rate the engine would apply in the absence of any PET forcing, including monthly adjustments, in user units (in/day for US projects, mm/day for SI). Intended for caller-side composition: read this rate, apply your own adjustment logic, and prescribe the result via forcing_set_forcing with target_type="subcatchment" and variable="evap".

async openswmm_mcp.tools.forcing.get_climate_state(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Read back the current climate inputs (read-only).

Returns the air temperature, wind speed, dry-only flag, and the climate-derived evaporation rate the engine is currently using (after any forcing). Requires the openswmm backend.

async openswmm_mcp.tools.forcing.set_climate_dry_only(ctx: fastmcp.Context, session_id: str = 'default', flag: bool = True) → dict#

Toggle the climate “evaporate only during dry weather” rule.

When enabled, evaporation is suppressed during rainfall periods. Requires the openswmm backend; takes effect on the next step.

Parameters:

session_id – Target simulation session.
flag – True suppresses evaporation during rainfall; False allows it.

async openswmm_mcp.tools.forcing.set_climate_forcing(ctx: fastmcp.Context, session_id: str = 'default', variable: str = '', value: float = 0.0, mode: str = 'replace', persist: bool = False) → dict#

Apply a model-global climate forcing override.

Overrides a climate input that applies to the whole model (not a single element): air temperature, wind speed, or potential evaporation. These feed snowmelt, evaporation, and other climate-driven processes from the next step on.

Climate forcing is a v1-only capability and requires the openswmm backend.

Parameters:

session_id – Target simulation session.
variable – "temperature" (air temperature, project units), "wind" (wind speed, project units), or "evap" (potential evaporation rate, in/day US or mm/day SI).
value – The forcing value to apply.
mode – "replace" (default) overwrites the climate-derived value; "add" adds to it.
persist – If True the override persists across timesteps; otherwise it resets after each step.

async openswmm_mcp.tools.forcing.set_forcing(ctx: fastmcp.Context, session_id: str = 'default', target_type: str = '', element_id: str = '', variable: str = '', value: float = 0.0, mode: str = 'replace', persist: bool = False) → ForcingResult#

Apply a runtime forcing override to a model element.

Overrides the value of a specific variable on a node, link, subcatchment, or rain gage for the current (and optionally future) timesteps.

Parameters:

session_id – Target simulation session.
target_type – Element category: "node", "link", "subcatchment", or "gage".
element_id – The name or index of the element to force.
variable – The variable to override. Valid choices depend on target_type: node ("lateral_inflow", "head", "quality"), link ("flow", "setting"), subcatchment ("rainfall", "evap", "snowfall"), gage ("rainfall").
value – The forcing value to apply.
mode – "replace" (default) overwrites the computed value; "add" adds to it.
persist – If True the override persists across timesteps; otherwise it resets after each step.

async openswmm_mcp.tools.forcing.set_link_control(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', setting: float = 0.0) → dict#

Set the control setting on a link.

Directly overrides a link’s control setting (e.g. pump speed, orifice opening fraction) for the current timestep.

Parameters:

session_id – Target simulation session.
link_id – The name or index of the link.
setting – The new control setting value.

async openswmm_mcp.tools.forcing.set_link_quality(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', pollutant: str = '', value: float = 0.0, mode: str = 'replace', persist: bool = False) → dict#

Force a pollutant concentration on a link (RUNNING state only).

Overrides the in-link concentration of a single pollutant for the current (and, with persist=True, future) timesteps. The element-keyed set_forcing() covers node quality but not link quality, so this is the dedicated link-quality forcing tool.

Link quality forcing is a v1-only capability and requires the openswmm backend.

Parameters:

session_id – Target simulation session.
link_id – The name or index of the link.
pollutant – The name or index of the pollutant.
value – The concentration to apply (model concentration units).
mode – "replace" (default) overwrites the computed value; "add" adds to it.
persist – If True the override persists across timesteps; otherwise it resets after each step.

async openswmm_mcp.tools.forcing.set_persistent_forcing(ctx: fastmcp.Context, session_id: str = 'default', target_type: str = '', element_id: str = '', variable: str = '', value: float = 0.0, mode: str = 'replace') → ForcingResult#

Apply a forcing override that persists across timesteps.

Equivalent to set_forcing() with persist=True, surfaced as its own tool so an LLM doesn’t have to know about the persist flag to get a sticky override. Use clear_forcing() to remove the override later.

Sticky overrides are a v1-only capability — they require the new engine. On the legacy backend the call is rejected with NOT_SUPPORTED because legacy resets API values every step automatically.

Parameters:

target_type – "node" / "link" / "subcatchment" / "gage".
element_id – Element ID (string) or numeric index as a string.
variable – Variable to override. Same set as set_forcing(): node lateral_inflow / head / quality; link flow / setting; subcatchment rainfall / evap; gage rainfall.
value – The forced value (units match the variable).
mode – "replace" (default) overwrites the computed value; "add" adds to it.

async openswmm_mcp.tools.forcing.set_rainfall_override(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', rainfall: float = 0.0) → dict#

Override rainfall on a rain gage with a persistent replacement value.

This is a convenience shortcut that applies a REPLACE + PERSIST forcing on the specified rain gage. Use clear_forcing() to remove the override later.

Parameters:

session_id – Target simulation session.
gage_id – The name or index of the rain gage.
rainfall – The rainfall intensity to apply (in the model’s rainfall units).

openswmm_mcp.tools.controls#

Controls tools: lifecycle-spanning control-rule management.

Wraps openswmm.engine.Controls for full-lifecycle access to the [CONTROLS] section. Complements (does not replace) the two runtime-only tools in openswmm_mcp.tools.forcing:

forcing.set_link_control — single-step pump/orifice/weir setting, state=running only.
forcing.add_control_rule — add a rule mid-simulation, state=running only.

The tools here cover the same C API surface but at design / setup time: read existing rules, list / inspect / clear them, and configure link behaviour outside of an active simulation step.

C-API state contract (per openswmm_controls.h):

swmm_control_add_rule, swmm_control_count, swmm_control_get_rule, swmm_control_clear_rules carry no state annotation and work in any non-closed state (building, opened, initialized, running, ended).
swmm_control_set_link_setting and swmm_control_set_link_status are documented as RUNNING state only.

async openswmm_mcp.tools.controls.add_rule(ctx: fastmcp.Context, session_id: str = 'default', rule_text: str = '') → dict#

Add a control rule to the model (lifecycle-spanning).

Accepts the full SWMM rule text including the RULE <id> header, one or more IF / AND / OR clauses, and a THEN action block (and optional ELSE / PRIORITY clauses). Lines are newline-separated within the string.

Works in any non-closed state. The runtime-only counterpart forcing.add_control_rule enforces state=running and is the right tool when adding rules mid-simulation.

Example

RULE PUMP_ON
IF NODE J1 DEPTH > 5.0
THEN PUMP P1 STATUS = ON

async openswmm_mcp.tools.controls.clear_rules(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Remove every control rule from the model.

async openswmm_mcp.tools.controls.count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of control rules defined in the model.

async openswmm_mcp.tools.controls.find_references(ctx: fastmcp.Context, session_id: str = 'default', object_name: str = '') → dict#

Return the indices of control rules that reference an object by name.

Wraps Controls.find_references. Scans each rule’s clauses for an object-type keyword (NODE / LINK / CONDUIT / PUMP / ORIFICE / WEIR / OUTLET) immediately followed by object_name (case-insensitive). Read-only — no rule text is edited. Use this before deleting or renaming an object to find the rules that would be affected.

async openswmm_mcp.tools.controls.get_id(ctx: fastmcp.Context, session_id: str = 'default', rule_index: int = 0) → dict#

Return the canonical rule name parsed from the I{rule_index}-th control rule’s text (the first token after the RULE keyword, case-insensitive).

When the rule text is malformed (no parseable RULE keyword token), name is None so callers can render a sentinel display label like Rule N [unnamed] without catching exceptions.

async openswmm_mcp.tools.controls.get_rule(ctx: fastmcp.Context, session_id: str = 'default', rule_index: int = 0) → dict#

Return the full text of the I{rule_index}-th control rule.

The rule text is multi-line: a RULE <id> header followed by IF / AND / OR clauses and a THEN action block.

async openswmm_mcp.tools.controls.list_rules(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return all control rules as a list of {index, name, text} dicts.

async openswmm_mcp.tools.controls.remove_rule(ctx: fastmcp.Context, session_id: str = 'default', rule_index: int = 0) → dict#

Remove a single control rule by index (later rules shift down by one).

Wraps Controls.remove_rule. Unlike clear_rules(), which drops every rule, this deletes only the I{rule_index}-th rule; all rules after it renumber down. Requires the engine to be in building or opened state.

async openswmm_mcp.tools.controls.set_link_setting(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', setting: float = 0.0) → dict#

Set a continuous control setting on a link (RUNNING state only).

Maps to swmm_control_set_link_setting. Used for pump speeds, orifice openings, weir crest positions — anywhere the engine model accepts a 0..1 (or higher, depending on link type) continuous value.

Distinct from set_link_status() which sets a discrete OPEN/CLOSED state. The plan documents the split: setting is continuous, status is binary.

For mid-simulation control, prefer forcing.set_link_control which is functionally equivalent — both wrap the same C call. This tool exists in the controls namespace for naming symmetry with the rest of the rule-management surface.

async openswmm_mcp.tools.controls.set_link_status(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', open: bool = True) → dict#

Set the discrete OPEN/CLOSED status of a link (RUNNING state only).

Maps to swmm_control_set_link_status. The boolean open is forwarded as the inverse to v1’s keyword-only closed argument.

For continuous control settings (pump speed, orifice opening), use set_link_setting().

async openswmm_mcp.tools.controls.validate_rule(ctx: fastmcp.Context, session_id: str = 'default', rule_text: str = '') → dict#

Validate control-rule text WITHOUT adding it to the model.

Parses rule_text through the engine’s rule compiler and reports whether it is syntactically valid. On failure message carries the engine’s diagnostic string; on success it is empty. Use this to pre-flight a rule before committing it via add_rule() (or the runtime forcing.add_control_rule).

Accepts the full SWMM rule text (the RULE <id> header, IF / AND / OR clauses, and a THEN action block). Works in any non-closed state and never mutates the model.

openswmm_mcp.tools.analysis#

Post-simulation analysis tools for the OpenSWMM MCP server.

Provides tools for querying simulation statistics, mass balance, time series output, flooding summaries, capacity summaries, scenario comparison, and result export.

async openswmm_mcp.tools.analysis.compare_scenarios(ctx: fastmcp.Context, session_a: str = '', session_b: str = '', element_type: str = 'node', variable: str = 'depth') → dict#

Compare time-series output between two simulation sessions.

Retrieves the specified variable for all elements of the given type from both sessions and returns summary statistics of the differences (mean, max, min of the element-wise peak differences).

Parameters:

session_a – Session identifier for the baseline scenario.
session_b – Session identifier for the comparison scenario.
element_type – One of "node", "link", or "subcatchment".
variable – The output variable to compare (e.g. "depth", "flow").

async openswmm_mcp.tools.analysis.export_results(ctx: fastmcp.Context, session_id: str = 'default', output_path: str = '', format: str = 'csv') → ExportResult#

Export node and link time-series results to CSV or JSON.

Writes one file containing all node and link output variables for every reporting period. Useful for downstream analysis in spreadsheets or data-science tools.

Parameters:

session_id – Identifier of the session. Defaults to "default".
output_path – Destination file path. Relative paths are resolved against the session’s working directory.
format – Output format: "csv" or "json". Defaults to "csv".

async openswmm_mcp.tools.analysis.get_capacity_summary(ctx: fastmcp.Context, session_id: str = 'default', max_filling_threshold: float = 1.0) → list[CapacitySummaryItem]#

Summarise hydraulic capacity usage across all links in the model.

Returns a list of links whose maximum depth-to-full-depth ratio exceeds max_filling_threshold, sorted by filling ratio in descending order.

Parameters:

session_id – Identifier of the session. Defaults to "default".
max_filling_threshold – Links with max_depth / full_depth above this value are included. Defaults to 1.0 (i.e. links that exceeded full capacity).

async openswmm_mcp.tools.analysis.get_flooding_summary(ctx: fastmcp.Context, session_id: str = 'default', min_flood_volume: float = 0.0) → list[FloodingSummaryItem]#

Summarise flooding across all nodes in the model.

Returns a list of nodes that experienced flooding (volume > min_flood_volume), sorted by total flood volume in descending order.

Parameters:

session_id – Identifier of the session. Defaults to "default".
min_flood_volume – Minimum flood volume threshold. Nodes with a total flood volume at or below this value are excluded. Defaults to 0.0.

async openswmm_mcp.tools.analysis.get_mass_balance(ctx: fastmcp.Context, session_id: str = 'default') → MassBalanceResult#

Retrieve mass-balance continuity errors and volumetric totals.

Returns the runoff, routing, and (if pollutants exist) quality continuity errors together with the individual volume components (rainfall, runoff, flooding, outflow, etc.).

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.analysis.get_pump_summary(ctx: fastmcp.Context, session_id: str = 'default') → list[PumpEntryModel]#

Return post-simulation performance statistics for all pump links.

Equivalent to the Pumping Summary section of the SWMM .rpt file. Only links of type PUMP are returned; if the model has no pumps the list will be empty.

Each entry includes:

link_id — pump identifier
pump_curve_idx — index of the pump curve used (-1 = ideal pump)
num_startups — total on/off cycles during the simulation
total_on_time — cumulative run time (seconds)
total_volume — total volume pumped
pct_time_on — percentage of simulation duration the pump was active

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.analysis.get_quality_losses(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return per-pollutant evaporation and seepage mass losses.

These are the quality mass-balance loss terms the routing continuity accounts for but which get_mass_balance() does not surface (it reports continuity errors and volumetric totals). For every modelled pollutant, returns the cumulative mass lost to evaporation and to seepage (model mass units).

Requires the openswmm backend and a session in running or ended state.

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.analysis.get_report_snapshot(ctx: fastmcp.Context, session_id: str = 'default') → ReportSnapshotModel#

Return the full post-simulation report as structured data.

Assembles the programmatic equivalent of the SWMM .rpt file into a single structured response covering:

Routing diagnostics — time-step statistics and convergence metrics (average/min/max step, total steps, number and percentage of non-converging steps, average iterations, maximum Courant number)
Runoff continuity — rainfall, evaporation, infiltration, runoff, and storage change volumes with continuity error
Flow routing continuity — inflow components, flooding, outflow, and loss volumes with continuity error
Quality continuity — per-pollutant mass balance with seep and evap losses (empty when no pollutants are modelled)
Node flooding summary — all nodes that experienced overflow, sorted by total flood volume
Storage volume summary — all STORAGE-type nodes with depth and volume statistics
Link flow summary — all links with peak flow, velocity, filling ratio, total volume, and surcharge time
Pump summary — PUMP links with startup count, total on-time, volume pumped, and percentage time on
Subcatchment runoff summary — precipitation, runoff volume, peak rate, and runoff coefficient per subcatchment

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.analysis.get_statistics(ctx: fastmcp.Context, session_id: str = 'default', element_type: str = 'node', element_id: str = '') → dict#

Retrieve post-simulation statistics for a single model element.

Returns peak / max values and duration statistics collected by the engine during the simulation run.

Parameters:

session_id – Identifier of the session. Defaults to "default".
element_type – One of "node", "link", or "subcatchment".
element_id – The identifier (name) of the element.

async openswmm_mcp.tools.analysis.get_time_series(ctx: fastmcp.Context, session_id: str = 'default', element_type: str = 'node', element_id: str = '', variable: str = 'depth', start_period: int = 0, end_period: int = -1, downsample: int = 1) → TimeSeries#

Retrieve a time series of output results for a model element.

Reads from the binary .out file produced by the simulation. The start_period and end_period parameters select a slice of the reporting periods (0-indexed). Use downsample to skip periods for large result sets (e.g. downsample=10 returns every 10th value).

Parameters:

session_id – Identifier of the session. Defaults to "default".
element_type – One of "node", "link", "subcatchment", or "system".
element_id – The identifier of the element. Ignored when element_type is "system".
variable – The output variable to retrieve (e.g. "depth", "flow").
start_period – First reporting period index (inclusive, 0-based). Defaults to 0.
end_period – Last reporting period index (exclusive). -1 means all periods.
downsample – Take every N-th value. Defaults to 1 (no downsampling).

async openswmm_mcp.tools.analysis.output_link_attribute(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', period: int = 0) → dict#

Return all variable values for a link at a single reporting period.

Variables are returned as a dict keyed by name: flow, depth, velocity, volume, capacity, plus pollutant_i columns when pollutants are tracked.

async openswmm_mcp.tools.analysis.output_link_results(ctx: fastmcp.Context, session_id: str = 'default', variable: str = 'flow', period: int = 0) → dict#

Return one link variable across all links at a single reporting period.

variable is one of: flow, depth, velocity, volume, capacity.

Returns a list of {id, index, value} records.

async openswmm_mcp.tools.analysis.output_metadata(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return header metadata for the .out file (counts + timing + version).

Combines several small reader getters into one call so an LLM can size a subsequent batch read in a single round-trip.

async openswmm_mcp.tools.analysis.output_node_attribute(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', period: int = 0) → dict#

Return all variable values for a node at a single reporting period.

Variables are returned as a dict keyed by name: depth, head, volume, lateral_inflow, total_inflow, overflow, plus pollutant_0 .. pollutant_{n-1} when pollutants are tracked.

Wraps swmm_output_get_node_attribute — the per-object snapshot accessor distinct from get_time_series (one variable over time) and get_node_result (one variable over all nodes at one period).

async openswmm_mcp.tools.analysis.output_node_results(ctx: fastmcp.Context, session_id: str = 'default', variable: str = 'depth', period: int = 0) → dict#

Return one node variable across all nodes at a single reporting period.

variable is one of: depth, head, volume, lateral_inflow, total_inflow, overflow.

Returns a list of {id, index, value} records ordered by node index (which matches the .out file’s stored order).

async openswmm_mcp.tools.analysis.output_node_stats(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '') → dict#

Return post-run node statistics aggregated from the .out file.

Wraps the four engine-level accessors swmm_output_get_node_stat_max_depth / _max_overflow / _vol_flooded / _time_flooded.

These differ from get_flooding_summary() in two ways:

the aggregation is computed from the binary output file (so it works even after the engine handle is closed); and

the response covers a single named node, not a filtered list across the whole network.

Parameters:

session_id – Identifier of the session. Must be in ended state.
node_id – String node identifier (e.g. "J1"). Required.

Returns:

{"session_id": ..., "node_id": ..., "node_index": ..., "max_depth": float, "max_overflow": float, "total_flood_volume": float, "time_flooded_seconds": float, "time_flooded_hours": float}. time_flooded_hours is provided as a convenience because that is the convention SWMM’s statsrpt uses for display.

Return type:

dict

Raises:

ToolError – Session not in ended state, backend is the legacy engine (this feature requires the new engine), node_id is empty, or the node does not exist.

async openswmm_mcp.tools.analysis.output_period_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of reporting periods written to the .out file.

async openswmm_mcp.tools.analysis.output_period_time(ctx: fastmcp.Context, session_id: str = 'default', period: int = 0) → dict#

Return the elapsed time (project time units) for a reporting period.

The value combines with start_date (from output_metadata()) to produce an absolute timestamp.

async openswmm_mcp.tools.analysis.output_pollutant_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of pollutants tracked in the .out file.

async openswmm_mcp.tools.analysis.output_subcatch_attribute(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', period: int = 0) → dict#

Return all variable values for a subcatchment at a reporting period.

Variables: rainfall, snow_depth, evap, infil, runoff, gw_flow, gw_elev, soil_moist, plus pollutant_i columns when pollutants are tracked.

async openswmm_mcp.tools.analysis.output_subcatch_results(ctx: fastmcp.Context, session_id: str = 'default', variable: str = 'runoff', period: int = 0) → dict#

Return one subcatchment variable across all subcatchments at a period.

variable is one of: rainfall, snow_depth, evap, infil, runoff, gw_flow, gw_elev, soil_moist.

Returns a list of {id, index, value} records.

async openswmm_mcp.tools.analysis.output_system_result(ctx: fastmcp.Context, session_id: str = 'default', variable: str = 'rainfall', period: int = 0) → dict#

Return a single system-level variable at a single reporting period.

Cheaper than get_time_series when only one timestep is needed. variable is one of: temperature, rainfall, snow_depth, evap, infil, runoff, dw_inflow, gw_inflow, lat_inflow, flooding, outflow, storage, evap_total, pet.

openswmm_mcp.tools.building#

Programmatic model-construction tools for the OpenSWMM MCP server.

Provides tools for creating SWMM models from scratch using the ModelBuilder API – adding nodes, links, subcatchments, gages, time series, curves, and validating / exporting the result.

async openswmm_mcp.tools.building.add_curve(ctx: fastmcp.Context, session_id: str = 'default', name: str = '', curve_type: str = 'storage', x_values: list[float] | None = None, y_values: list[float] | None = None) → dict#

Add a curve to the model.

Parameters:

name – Unique name for the curve.
curve_type – Curve type: storage, pump, rating, diversion, tidal, shape, weir, control.
x_values – List of x-axis values.
y_values – List of corresponding y-axis values (same length as x_values).

async openswmm_mcp.tools.building.add_gage(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '') → BuildingResult#

Add a rain gage to the model.

Parameters:: gage_id – Unique identifier for the rain gage.

async openswmm_mcp.tools.building.add_link(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', link_type: str = 'conduit', from_node: str = '', to_node: str = '', length: float = 100.0, roughness: float = 0.013, xsect_shape: str = 'circular', xsect_geom1: float = 1.0, xsect_geom2: float = 0.0, xsect_geom3: float = 0.0, xsect_geom4: float = 0.0) → BuildingResult#

Add a link (conduit, pump, orifice, weir, or outlet) to the model.

Parameters:

link_id – Unique identifier for the link.
link_type – One of conduit, pump, orifice, weir, outlet.
from_node – IDs of the upstream and downstream nodes.
to_node – IDs of the upstream and downstream nodes.
length – Conduit length (ft or m).
roughness – Manning’s roughness coefficient.
xsect_shape – Cross-section shape: circular, rect_closed, rect_open, trapezoidal, triangular.
xsect_geom4 (xsect_geom1 ..) – Shape-dependent geometry parameters (e.g. diameter for circular).

async openswmm_mcp.tools.building.add_node(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', node_type: str = 'junction', invert_elev: float = 0.0, max_depth: float = 0.0, x: float | None = None, y: float | None = None) → BuildingResult#

Add a node to the model being built.

Parameters:

node_id – Unique identifier for the node.
node_type – One of junction, outfall, storage, divider.
invert_elev – Invert elevation (ft or m depending on flow units).
max_depth – Maximum depth above invert (0 = use default).
x – Optional coordinate position for spatial display.
y – Optional coordinate position for spatial display.

async openswmm_mcp.tools.building.add_pollutant(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str = '', units: str = 'mg/l', kdecay: float = 0.0, rain_conc: float = 0.0, gw_conc: float = 0.0, init_conc: float = 0.0, snow_only: bool = False) → BuildingResult#

Add a pollutant to the model.

Valid in building or opened state. After adding, the pollutant can be referenced by its ID when configuring buildup/washoff or quality injection.

Parameters:

pollutant_id – Unique pollutant identifier (e.g. TSS, TN).
units – Concentration units: mg/l, ug/l, or #/l.
kdecay – First-order decay coefficient (1/days).
rain_conc – Concentration in rainfall (same units as pollutant).
gw_conc – Concentration in groundwater inflow.
init_conc – Initial concentration in the network.
snow_only – If True, buildup occurs only during snow accumulation.

async openswmm_mcp.tools.building.add_subcatchment(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', area: float = 1.0, imperv_pct: float = 50.0, slope: float = 0.5, width: float = 100.0, outlet_node: str = '') → BuildingResult#

Add a subcatchment to the model.

Parameters:

subcatch_id – Unique identifier for the subcatchment.
area – Subcatchment area (acres or hectares).
imperv_pct – Percent imperviousness (0-100).
slope – Average surface slope (percent).
width – Characteristic width for overland flow (ft or m).
outlet_node – ID of the node (or another subcatchment) receiving runoff.

async openswmm_mcp.tools.building.add_timeseries(ctx: fastmcp.Context, session_id: str = 'default', name: str = '', times: list[float] | None = None, values: list[float] | None = None) → dict#

Add a time series to the model.

Parameters:

name – Unique name for the time series.
times – List of time values (hours from simulation start).
values – List of corresponding data values (same length as times).

async openswmm_mcp.tools.building.create_model(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Create an empty SWMM model and start a building session.

Returns metadata for the new session. Use the add_node, add_link, add_subcatchment, and related tools to populate the model before calling validate_model or write_model.

async openswmm_mcp.tools.building.pop_last_link(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '') → BuildingResult#

Remove the most recently added link (undo of add_link).

The supplied link_id must match the current tail of the link list, otherwise the engine returns SWMM_ERR_BADINDEX.

Parameters:: link_id – Expected tail link identifier.

async openswmm_mcp.tools.building.pop_last_node(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '') → BuildingResult#

Remove the most recently added node (undo of add_node).

The supplied node_id must match the current tail of the node list. If any link still references the tail node, the engine refuses the pop — call pop_last_link for those links first.

Parameters:: node_id – Expected tail node identifier.

async openswmm_mcp.tools.building.set_option(ctx: fastmcp.Context, session_id: str = 'default', option: str = '', value: str = '') → dict#

Set a simulation option on the model being built.

Parameters:

option – Option name (e.g. FLOW_UNITS, ROUTING_MODEL, REPORT_STEP).
value – Option value as a string.

async openswmm_mcp.tools.building.validate_model(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Validate the model being built.

Runs the engine’s built-in validation checks and returns any warnings or errors. A model with no messages is considered valid.

async openswmm_mcp.tools.building.write_model(ctx: fastmcp.Context, session_id: str = 'default', output_path: str = '') → dict#

Finalize and write the model to an .inp file.

If the session is still in building state, the ModelBuilder is finalized to produce a Solver, which is then used to write the file. If the session already has a solver (e.g. it was previously finalized), the existing solver writes the file directly.

Parameters:: output_path – Filesystem path for the output .inp file.

openswmm_mcp.tools.editing#

Model-editing tools — object deletion, type conversion, and property updates.

These tools operate on sessions in building (programmatic construction) or opened (after parsing a .inp file) state. They expose the ModelEditor C API surface: non-destructive impact analysis, cascade deletion, in-place node / link type conversion, and in-place property updates.

async openswmm_mcp.tools.editing.analyze_impact(ctx: fastmcp.Context, session_id: str = 'default', object_type: str = 'node', object_id: str = '') → ImpactReportModel#

Preview what would be affected if an object were deleted, without deleting it.

Use this to inspect cascades and reference nullifications before calling delete_object. No objects are modified.

Parameters:

object_type – One of node, link, subcatchment, gage, table, transect, pollutant, pattern, aquifer, snowpack, lid, street, inlet, landuse, hydrograph.
object_id – The object’s string identifier (or a numeric index as a string for transect).

async openswmm_mcp.tools.editing.configure_gage(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', rain_type: str | None = None, rain_interval: float | None = None, data_source: str | None = None, timeseries_id: str | None = None, filename: str | None = None, station_id: str | None = None) → GageConfigResult#

Configure a rain gage’s data source and recording parameters.

Only fields that are explicitly provided (non-null) are updated. Valid in building, opened, or initialized state.

Parameters:

gage_id – Gage identifier.
rain_type – Rainfall measurement type: intensity, volume, or cumulative.
rain_interval – Recording interval in seconds.
data_source – Data source type: timeseries or file.
timeseries_id – ID of the time-series table to use (when data_source is timeseries).
filename – Path to an external rainfall data file (when data_source is file).
station_id – Station identifier within the external file.

async openswmm_mcp.tools.editing.convert_link(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', new_type: str = 'conduit') → ConversionResultModel#

Convert a link to a different type in place.

Common properties (endpoint nodes, offsets, initial flow) are preserved. Type-specific properties are cleared and new-type defaults applied.

Parameters:

link_id – Link identifier or zero-based index.
new_type – Target type: conduit, pump, orifice, weir, or outlet.

async openswmm_mcp.tools.editing.convert_node(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', new_type: str = 'junction') → ConversionResultModel#

Convert a node to a different type in place.

Common properties (invert elevation, max depth, coordinates) are preserved. Type-specific properties for the old type are cleared and sensible defaults for the new type are applied. Non-fatal topology warnings are reported but do not prevent conversion.

Parameters:

node_id – Node identifier or zero-based index.
new_type – Target type: junction, outfall, storage, or divider.

async openswmm_mcp.tools.editing.delete_object(ctx: fastmcp.Context, session_id: str = 'default', object_type: str = 'node', object_id: str = '', dry_run: bool = False) → ImpactReportModel#

Delete a model object and cascade-delete or nullify all referencing objects.

When dry_run is True the impact is analysed but nothing is deleted (equivalent to analyze_impact()).

Cascade policy

Links that reference a deleted node as an endpoint are deleted.
Subcatchment outlet_node, inlet-usage node_index, and similar weak references are nullified (set to -1).
All integer cross-references whose value exceeded the deleted index are decremented by 1.

Parameters:

object_type – One of node, link, subcatchment, gage, table, transect, pollutant, pattern, aquifer, snowpack, lid, street, inlet, landuse, hydrograph.
object_id – String identifier of the object to delete, or a numeric index string for transect.
dry_run – When True, return the impact report without mutating the model.

async openswmm_mcp.tools.editing.get_gage_metadata(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '') → dict#

Read back the data-source metadata configure_gage() writes.

configure_gage is write-only for these fields and query_get_gage_info does not return them, so this is the only way to verify what a gage was configured with:

rain_interval — recording interval in seconds.
rain_units — in / mm, the depth unit declared for a file source. Distinct from rain_type (intensity / volume / cumulative), which query_get_gage_info already reports.
timeseries_id — assigned series id (empty for a file source).
station_id — station id within an external file (empty otherwise).

Valid in building, opened, or initialized state.

Parameters:: gage_id – Gage identifier.

async openswmm_mcp.tools.editing.get_gage_scale_factor(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '') → dict#

Return a rain gage’s rainfall scale factor.

The scale factor multiplies the gage’s raw rainfall series — values above 1.0 amplify, below 1.0 attenuate. configure_gage() does not touch it; use set_gage_scale_factor() to change it. Valid in building, opened, or initialized state.

Parameters:: gage_id – Gage identifier.

async openswmm_mcp.tools.editing.get_gage_snow_factor(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '') → dict#

Return a rain gage’s snow catch factor (SCF).

The SCF corrects the physical gage’s snow-catch deficiency: below the snow temperature threshold, snowfall is multiplied by it. Distinct from the rainfall get_gage_scale_factor() — SCF affects only the snow branch. Valid in building, opened, or initialized state.

Parameters:: gage_id – Gage identifier.

async openswmm_mcp.tools.editing.get_subcatch_rain_scale_factor(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '') → dict#

Return a subcatchment’s rainfall scale factor.

Optional [SUBCATCHMENTS] token 9 (default 1.0). Multiplies this subcatchment’s gage-derived rainfall only, composing with the gage’s own scale factor. Valid in building, opened, or initialized state.

Parameters:: subcatch_id – Subcatchment identifier.

async openswmm_mcp.tools.editing.get_subcatch_snow_scale_factor(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '') → dict#

Return a subcatchment’s snowfall scale factor.

Optional [SUBCATCHMENTS] token 10 (default 1.0). Composes with the gage snow catch factor (SCF). Valid in building, opened, or initialized state.

Parameters:: subcatch_id – Subcatchment identifier.

async openswmm_mcp.tools.editing.rename_gage(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', new_id: str = '') → dict#: Rename a rain gage (wraps swmm_gage_rename).

async openswmm_mcp.tools.editing.rename_landuse(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str = '', new_id: str = '') → dict#

Rename a land use.

Land uses are referenced positionally, so buildup / washoff rows and subcatchment coverages follow automatically.

async openswmm_mcp.tools.editing.rename_link(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', new_id: str = '') → dict#: Rename a link (wraps swmm_link_rename).

async openswmm_mcp.tools.editing.rename_node(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', new_id: str = '') → dict#

Rename a node (wraps swmm_node_rename).

The new id must be unique across the node namespace and non-empty. The engine returns SWMM_ERR_BADPARAM on collision or empty input.

async openswmm_mcp.tools.editing.rename_pattern(ctx: fastmcp.Context, session_id: str = 'default', pattern_id: str = '', new_id: str = '') → dict#: Rename a time pattern, updating every stored reference to it.

async openswmm_mcp.tools.editing.rename_pollutant(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str = '', new_id: str = '') → dict#

Rename a pollutant.

Name-stored references ([INFLOWS] / [DWF] constituent rows) follow the new name; index-stored ones (co-pollutant, buildup / washoff columns) are positional and unaffected.

async openswmm_mcp.tools.editing.rename_subcatchment(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', new_id: str = '') → dict#: Rename a subcatchment (wraps swmm_subcatch_rename).

async openswmm_mcp.tools.editing.rename_transect(ctx: fastmcp.Context, session_id: str = 'default', transect_id: str | int = '', new_id: str = '') → dict#: Rename a transect (by id or zero-based index), updating stored references.

async openswmm_mcp.tools.editing.set_gage_rain_units(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', rain_units: str = 'in') → dict#

Set the rain-depth units declared for a file-based gage.

rain_units is in (inches) or mm (millimetres). This is the depth unit of the values in the external file — not the rain type (intensity / volume / cumulative), which configure_gage() sets. configure_gage() does not touch it. Valid in building, opened, or initialized state.

Parameters:

gage_id – Gage identifier.
rain_units – in or mm.

async openswmm_mcp.tools.editing.set_gage_scale_factor(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', scale_factor: float = 1.0) → dict#

Set a rain gage’s rainfall scale factor.

The scale factor multiplies the gage’s raw rainfall series. Maps to the v1 Gage.scale_factor attribute, which configure_gage() leaves untouched. Valid in building, opened, or initialized state.

Parameters:

gage_id – Gage identifier.
scale_factor – Multiplier applied to the gage’s rainfall (1.0 = unchanged).

async openswmm_mcp.tools.editing.set_gage_snow_factor(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', snow_factor: float = 1.0) → dict#

Set a rain gage’s snow catch factor (SCF; must be > 0).

The SCF multiplies the gage’s snowfall (the below-freezing branch of the rain/snow split); it does not touch rainfall. Maps to the Gage.snow_factor attribute, distinct from set_gage_scale_factor(). Valid in building, opened, or initialized state.

Parameters:

gage_id – Gage identifier.
snow_factor – Multiplier applied to the gage’s snowfall (1.0 = unchanged).

async openswmm_mcp.tools.editing.set_gage_station_id(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', station_id: str = '') → dict#

Set the station id a file-based gage reads from.

configure_gage() only applies station_id alongside a filename; this sets it on its own, e.g. to point an already-configured file source at a different station. Valid in building, opened, or initialized state.

Parameters:

gage_id – Gage identifier.
station_id – Station identifier within the external rainfall file.

async openswmm_mcp.tools.editing.set_link_properties(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', length: float | None = None, roughness: float | None = None, offset_up: float | None = None, offset_dn: float | None = None, initial_flow: float | None = None, max_flow: float | None = None, xsect_shape: str | None = None, xsect_geom1: float | None = None, xsect_geom2: float | None = None, xsect_geom3: float | None = None, xsect_geom4: float | None = None) → PropertyUpdateResult#

Update geometry properties of an existing link in place.

Only fields that are explicitly provided (non-null) are updated. Valid in building, opened, or initialized state.

Cross-section fields (xsect_shape, xsect_geom1–xsect_geom4) are applied as a group only when xsect_shape is provided.

Parameters:

link_id – Link identifier or zero-based index string.
length – Conduit length (project length units).
roughness – Manning’s roughness coefficient.
offset_up – Upstream invert offset above connecting node invert.
offset_dn – Downstream invert offset above connecting node invert.
initial_flow – Initial flow rate at simulation start.
max_flow – Maximum allowable flow rate (0 = no limit).
xsect_shape – Cross-section shape name (e.g. circular, rect_closed).
xsect_geom1–4 – Shape geometry parameters (meaning depends on shape type).

async openswmm_mcp.tools.editing.set_node_properties(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', invert_elev: float | None = None, max_depth: float | None = None, initial_depth: float | None = None, surcharge_depth: float | None = None, ponded_area: float | None = None) → PropertyUpdateResult#

Update geometry properties of an existing node in place.

Only fields that are explicitly provided (non-null) are updated. All others are left unchanged. Valid in building, opened, or initialized state.

Parameters:

node_id – Node identifier or zero-based index string.
invert_elev – Node invert elevation (project length units).
max_depth – Maximum node depth (project length units).
initial_depth – Initial water depth at simulation start.
surcharge_depth – Surcharge depth above the crown.
ponded_area – Ponded surface area when node is flooded.

async openswmm_mcp.tools.editing.set_subcatch_rain_scale_factor(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', scale_factor: float = 1.0) → dict#

Set a subcatchment’s rainfall scale factor (must be > 0).

Optional [SUBCATCHMENTS] token 9. Settable mid-run for parameter sweeps / RTC. Valid in building, opened, or initialized state.

Parameters:

subcatch_id – Subcatchment identifier.
scale_factor – Multiplier applied to this subcatchment’s rainfall (1.0 = unchanged).

async openswmm_mcp.tools.editing.set_subcatch_snow_scale_factor(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', scale_factor: float = 1.0) → dict#

Set a subcatchment’s snowfall scale factor (must be > 0).

Optional [SUBCATCHMENTS] token 10. Composes with the gage snow catch factor (SCF); settable mid-run. Valid in building, opened, or initialized state.

Parameters:

subcatch_id – Subcatchment identifier.
scale_factor – Multiplier applied to this subcatchment’s snowfall (1.0 = unchanged).

async openswmm_mcp.tools.editing.set_subcatchment_properties(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', area: float | None = None, width: float | None = None, slope: float | None = None, imperv_pct: float | None = None, n_imperv: float | None = None, n_perv: float | None = None, ds_imperv: float | None = None, ds_perv: float | None = None, outlet_node_id: str | None = None, gage_id: str | None = None) → PropertyUpdateResult#

Update properties of an existing subcatchment in place.

Only fields that are explicitly provided (non-null) are updated. Valid in building, opened, or initialized state.

Parameters:

subcatch_id – Subcatchment identifier.
area – Total subcatchment area (project area units).
width – Characteristic overland flow width (project length units).
slope – Average surface slope (fraction, e.g. 0.01 for 1%).
imperv_pct – Percent imperviousness (0–100).
n_imperv – Manning’s roughness for impervious area.
n_perv – Manning’s roughness for pervious area.
ds_imperv – Depression storage depth for impervious area.
ds_perv – Depression storage depth for pervious area.
outlet_node_id – ID of the node that receives runoff from this subcatchment.
gage_id – ID of the rain gage that drives this subcatchment.

openswmm_mcp.tools.model#

Model tools: title / userflag / options / CRS access.

The Python ModelBuilder (BUILDING state) and Solver (OPENED state and later) both expose:

[TITLE] section accessors: get_title_count / get_title_line / add_title_line / set_title / clear_title.
SWMM options: get_option / set_option (string-keyed) and get_option_ext / set_option_ext for extension options.
User flags: get/set_userflag_{bool,int,real}, application-defined metadata persisted alongside the model.
CRS string: get_crs.

This module surfaces all of those via the model MCP namespace. State handling: works against the ModelBuilder for BUILDING sessions, against the Solver for OPENED / INITIALIZED / RUNNING / ENDED.

async openswmm_mcp.tools.model.add_title_line(ctx: fastmcp.Context, session_id: str = 'default', text: str = '') → dict#: Append a line to the [TITLE] section.

async openswmm_mcp.tools.model.clear_title(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Remove every line from the [TITLE] section (BUILDING).

async openswmm_mcp.tools.model.file_path_get(ctx: fastmcp.Context, session_id: str = 'default', role: str = '', owner: str = '') → dict#

Read an external-file slot’s resolved and original paths.

role selects the slot: scalar roles RAINFALL, RUNOFF, RDII, INFLOWS, OUTFLOWS, HOTSTART_USE, CLIMATE_TEMP (owner ignored), or vector roles HOTSTART_SAVE (owner = decimal index), RAINGAGE_DATA (owner = gage id), TIMESERIES_DATA (owner = series id). Returns both the engine-resolved absolute path and the original token as authored in the .inp; either may be empty.

async openswmm_mcp.tools.model.file_path_set(ctx: fastmcp.Context, session_id: str = 'default', role: str = '', new_path: str = '', owner: str = '') → dict#

Set the original token for an external-file slot.

Clears the cached absolute resolution (the engine re-resolves on next use). For vector roles the owner must already exist in the model. Pass an empty new_path to clear the slot. See model_file_path_get for the role list.

async openswmm_mcp.tools.model.files_get(ctx: fastmcp.Context, session_id: str = 'default', key: str = '') → dict#

Return the path / value for a [FILES] section field.

Common keys: RAINFALL_PATH, RUNOFF_PATH, RDII_PATH, HOTSTART_USE_PATH, HOTSTART_SAVE_PATH.

async openswmm_mcp.tools.model.files_set(ctx: fastmcp.Context, session_id: str = 'default', key: str = '', value: str = '') → dict#: Set a [FILES] section field. Empty value clears the field.

async openswmm_mcp.tools.model.get_crs(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the model’s coordinate reference system string.

async openswmm_mcp.tools.model.get_option(ctx: fastmcp.Context, session_id: str = 'default', key: str = '') → dict#

Return a SWMM option value as a string.

Example keys: FLOW_UNITS, FLOW_ROUTING, ROUTING_STEP, REPORT_STEP, SURCHARGE_METHOD. Consult the SWMM 5 reference for the full key list.

async openswmm_mcp.tools.model.get_option_ext(ctx: fastmcp.Context, session_id: str = 'default', key: str = '') → dict#: Return an extension option value (unknown to base SWMM).

async openswmm_mcp.tools.model.get_pattern_factors(ctx: fastmcp.Context, session_id: str = 'default', pattern_id: str = '') → dict#

Read a time pattern’s type and multiplier factors.

Surfaces solver.patterns[...] — the multiplier list whose length depends on the pattern type (12 monthly, 7 daily, 24 hourly/weekend).

Parameters:: pattern_id – The [PATTERNS] id to read (required).

async openswmm_mcp.tools.model.get_report_start(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the report start date/time as an ISO 8601 string.

Surfaces the report_start_datetime property (present on both ModelBuilder and Solver). The report start is the instant from which reported results begin; it may lag the simulation start.

async openswmm_mcp.tools.model.get_title(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the full [TITLE] section as a list of lines (BUILDING).

Convenience wrapper that batches get_title_count + N x get_title_line.

async openswmm_mcp.tools.model.get_title_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of lines in the C{[TITLE]} section (BUILDING).

async openswmm_mcp.tools.model.get_title_line(ctx: fastmcp.Context, session_id: str = 'default', line_index: int = 0) → dict#: Return the I{line_index}-th line of the [TITLE] section (BUILDING).

async openswmm_mcp.tools.model.get_unit_system(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Report the model’s flow units and unit system.

Because the engine returns every quantity in the units declared in the .inp file (project units), a client must know those units to interpret returned magnitudes. This tool resolves [OPTIONS] FLOW_UNITS and classifies it:

flow_units — the raw token, e.g. "CFS" / "CMS".
unit_system — "US" (CFS/GPM/MGD) or "SI" (CMS/LPS/MLD).

Works in BUILDING (ModelBuilder) and OPENED/RUNNING/ENDED (Solver) states.

async openswmm_mcp.tools.model.get_userflag_bool(ctx: fastmcp.Context, session_id: str = 'default', name: str = '') → dict#: Return a boolean user flag (application-defined metadata).

async openswmm_mcp.tools.model.get_userflag_int(ctx: fastmcp.Context, session_id: str = 'default', name: str = '') → dict#: Return an integer user flag.

async openswmm_mcp.tools.model.get_userflag_real(ctx: fastmcp.Context, session_id: str = 'default', name: str = '') → dict#: Return a real-valued user flag.

async openswmm_mcp.tools.model.list_aquifers(ctx: fastmcp.Context, session_id: str = 'default') → dict#

List the model’s [AQUIFERS] entries.

Returns count and the ordered list of aquifer ids.

async openswmm_mcp.tools.model.list_snowpacks(ctx: fastmcp.Context, session_id: str = 'default') → dict#

List the model’s [SNOWPACKS] entries.

Returns count and the ordered list of snowpack ids.

async openswmm_mcp.tools.model.plugin_get(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the (path, args) of the I{index}-th plugin entry.

async openswmm_mcp.tools.model.plugin_remove(ctx: fastmcp.Context, session_id: str = 'default', path_or_id: str = '') → dict#: Remove the plugin entry matching path_or_id.

async openswmm_mcp.tools.model.plugin_set(ctx: fastmcp.Context, session_id: str = 'default', path_or_id: str = '', args: str = '') → dict#

Add or update a plugin entry.

path_or_id is the library path, plugin id, or id:version string.

async openswmm_mcp.tools.model.plugins_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of [PLUGINS] entries on the engine.

async openswmm_mcp.tools.model.set_option(ctx: fastmcp.Context, session_id: str = 'default', key: str = '', value: str = '') → dict#

Set a SWMM option (string key, string value).

Accepts any key the engine’s option API recognizes, including the FV_* family that configures the explicit finite-volume solver (FLOW_ROUTING = FV): FV_CELL_LENGTH, FV_MIN_CELLS, FV_CFL, FV_RIEMANN, FV_ORDER, FV_LIMITER, FV_SCALAR_SCHEME, FV_TIME_INTEGRATION, FV_SLOT_CELERITY, FV_DISPERSION, FV_STRUCTURE_COUPLING, FV_COMPACTION, FV_BACKEND and FV_MIN_PARALLEL_CELLS. These are inert under the other routing models rather than rejected, so they can be set before FLOW_ROUTING is switched.

Note that finite-volume routing needs a resolved mesh to reproduce dynamic-wave peak flows – set FV_CELL_LENGTH rather than leaving it at the one-cell-per-conduit default when peaks matter.

async openswmm_mcp.tools.model.set_option_ext(ctx: fastmcp.Context, session_id: str = 'default', key: str = '', value: str = '') → dict#: Set an extension option.

async openswmm_mcp.tools.model.set_report_start(ctx: fastmcp.Context, session_id: str = 'default', report_start: str = '') → dict#

Set the report start date/time from an ISO 8601 string.

report_start is parsed with datetime.datetime.fromisoformat() (e.g. "1998-01-01T00:00:00" or "1998-01-01 00:00:00").

async openswmm_mcp.tools.model.set_title(ctx: fastmcp.Context, session_id: str = 'default', text: str = '') → dict#: Replace all [TITLE] lines with new text (newline-separated, BUILDING).

async openswmm_mcp.tools.model.set_userflag_bool(ctx: fastmcp.Context, session_id: str = 'default', name: str = '', value: bool = False) → dict#: Set a boolean user flag.

async openswmm_mcp.tools.model.set_userflag_int(ctx: fastmcp.Context, session_id: str = 'default', name: str = '', value: int = 0) → dict#: Set an integer user flag.

async openswmm_mcp.tools.model.set_userflag_real(ctx: fastmcp.Context, session_id: str = 'default', name: str = '', value: float = 0.0) → dict#: Set a real-valued user flag.

async openswmm_mcp.tools.model.userflag_clear_value(ctx: fastmcp.Context, session_id: str = 'default', obj_type: str = '', obj_name: str = '', flag_name: str = '') → dict#: Remove the flag value assigned to a specific object (idempotent).

async openswmm_mcp.tools.model.userflag_define(ctx: fastmcp.Context, session_id: str = 'default', name: str = '', flag_type: str = '', description: str = '') → dict#

Define (or redefine) a user-flag schema entry ([USER_FLAGS]).

flag_type is "BOOLEAN", "INTEGER", "REAL", or "STRING". The name is stored uppercase. Redefining an existing name overwrites its definition; previously assigned per-object values are kept as-is.

async openswmm_mcp.tools.model.userflag_get_value(ctx: fastmcp.Context, session_id: str = 'default', obj_type: str = '', obj_name: str = '', flag_name: str = '') → dict#

Return the flag value assigned to a specific object ([USER_FLAG_VALUES]).

obj_type is an object type token (e.g. "NODE", "LINK", "SUBCATCHMENT"). The value is returned in its INP string form (BOOLEAN as YES/NO, INTEGER/REAL as decimals, STRING verbatim); value is None and assigned is False when unset.

async openswmm_mcp.tools.model.userflag_list_defs(ctx: fastmcp.Context, session_id: str = 'default') → dict#

List every user-flag schema definition ([USER_FLAGS]), in insertion order.

Each entry reports name, flag_type (BOOLEAN / INTEGER / REAL / STRING), and description.

async openswmm_mcp.tools.model.userflag_set_value(ctx: fastmcp.Context, session_id: str = 'default', obj_type: str = '', obj_name: str = '', flag_name: str = '', value: str = '') → dict#

Assign a flag value to a specific object from a string.

The flag must already be defined (see model_userflag_define); its declared type drives parsing. BOOLEAN accepts YES/NO/TRUE/FALSE/1/0; INTEGER a decimal integer; REAL a decimal number; STRING is stored verbatim.

async openswmm_mcp.tools.model.userflag_undefine(ctx: fastmcp.Context, session_id: str = 'default', name: str = '') → dict#: Remove a user-flag definition and all per-object values assigned to it.

async openswmm_mcp.tools.model.write_with_plugin(ctx: fastmcp.Context, session_id: str = 'default', path: str = '', output_plugin_id: str = '') → dict#

Write the model to disk via an output plugin (or built-in writer).

Pass an empty output_plugin_id (the default) to use the built-in .inp writer. Non-empty values select a registered output plugin (e.g. GeoPackage / HDF5).

openswmm_mcp.tools.nodes#

Nodes tools: fine-grained accessors beyond query.get_node_info.

The query.get_node_info() tool already returns a batched property dict for any node. This module adds individual tools for use cases the aggregate getter doesn’t serve well:

Statistics — post-simulation peak / duration metrics (4 tools).
Bulk array accessors — read/write all-node arrays in a single call (7 tools). Critical for visualization and batched edits.
Storage-node subtype — curve / functional / seep / exfiltration / shape+geometry (10 tools, getter+setter pairs).
Virtual junctions — is_virtual predicate and the virtual_eligible dry-run rule check (2 tools). Writing the flag goes through the editing tools.
Outfall-node subtype — type / stage / tidal / timeseries / flap_gate / route_to (8 tools).
Divider-node subtype — type get/set (2 tools).
Quality — point-wise pollutant getter + mass-flux injection setter (2 tools).
Conversion — depth-from-volume inverse lookup (1 tool).

Tools accept either an integer node index or a string node ID; resolution is done via session.nodes.get_index so unknown IDs surface as ELEMENT_NOT_FOUND.

The basic property pairs (invert_elev / max_depth / depth / head / volume / lateral_inflow / inflow / outflow / overflow / losses / degree / full_volume / crown_elev / initial_depth / surcharge_depth / ponded_area / type) are intentionally NOT duplicated here — those are already covered by query.get_node_info (read) and editing.set_node_properties (write).

async openswmm_mcp.tools.nodes.depth_from_volume(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', volume: float = 0.0) → dict#

Compute the depth corresponding to a given storage volume.

Inverts the storage curve / functional relationship for a storage node.

async openswmm_mcp.tools.nodes.get_depths_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current depths for all nodes as a list of {id, index, value}.

async openswmm_mcp.tools.nodes.get_divider_type(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return the divider rule type for a divider node.

divider_type enum: 0=CUTOFF, 1=OVERFLOW, 2=TABULAR, 3=WEIR.

async openswmm_mcp.tools.nodes.get_exfil_params(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return Green-Ampt exfiltration params (suction, ksat, imd) for a storage node.

async openswmm_mcp.tools.nodes.get_heads_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current heads for all nodes as a list of {id, index, value}.

async openswmm_mcp.tools.nodes.get_ids_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the IDs of all nodes in storage order as {count, ids}.

async openswmm_mcp.tools.nodes.get_inflows_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current total inflows for all nodes.

async openswmm_mcp.tools.nodes.get_outfall_flap_gate(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return whether an outfall has a flap gate (prevents backflow).

async openswmm_mcp.tools.nodes.get_outfall_param(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return the outfall parameter value (fixed stage or computed param).

Meaning depends on the outfall type: for FIXED it’s the stage elevation; for TIDAL / TIMESERIES it’s an index into a curve / series.

async openswmm_mcp.tools.nodes.get_outfall_route_to(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the subcatchment index outfall discharge is routed to (-1 = none).

async openswmm_mcp.tools.nodes.get_outfall_tidal(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return the tidal-curve index assigned to a TIDAL outfall.

Read-back of the curve set via set_outfall_tidal.

async openswmm_mcp.tools.nodes.get_outfall_timeseries(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return the stage-time-series index assigned to a TIMESERIES outfall.

Read-back of the series set via set_outfall_timeseries.

async openswmm_mcp.tools.nodes.get_outfall_type(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return the outfall boundary type code for an outfall node.

outfall_type enum: 0=FREE, 1=NORMAL, 2=FIXED, 3=TIDAL, 4=TIMESERIES.

async openswmm_mcp.tools.nodes.get_overflows_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current overflow rates for all nodes.

async openswmm_mcp.tools.nodes.get_quality(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', pollutant_index: int = 0) → dict#: Return the current concentration of a pollutant at a node.

async openswmm_mcp.tools.nodes.get_quality_bulk(ctx: fastmcp.Context, session_id: str = 'default', pollutant_index: int = 0) → dict#

Return pollutant concentrations at all nodes for one pollutant.

pollutant_index is a 0-based pollutant index (see query.get_pollutant_info or analysis.output_pollutant_count).

async openswmm_mcp.tools.nodes.get_storage_curve(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the storage-curve index assigned to a storage node.

async openswmm_mcp.tools.nodes.get_storage_functional(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return functional storage params (a, b, c) for a storage node.

Functional form: area = a * depth^b + c.

async openswmm_mcp.tools.nodes.get_storage_geometry(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return a storage node’s surface-area relation and its raw dimensions.

shape is the engine’s StorageShape: tabular (curve, see get_storage_curve), functional (a/b/c, see get_storage_functional), or one of the four geometric shapes, whose three raw dimensions p1/p2/p3 this tool returns:

cylindrical — p1 = major axis, p2 = minor axis.
conical — p1, p2 = base axes, p3 = side slope.
paraboloid — p1, p2 = top axes, p3 = height.
pyramidal — p1 = length, p2 = width, p3 = side slope.

p1/p2/p3 are zero for a non-geometric shape.

async openswmm_mcp.tools.nodes.get_storage_seep_rate(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the seepage rate for a storage node (depth/time).

async openswmm_mcp.tools.nodes.get_tag(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Return the free-form tag string for a node (empty if untagged).

Tags come from the INP [TAGS] section, are keyed by index, and persist across rename.

async openswmm_mcp.tools.nodes.is_virtual(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Report whether a node is a virtual junction.

A virtual junction is a zero-storage, momentum-transmitting JUNCTION connecting exactly two conduits of identical cross-section (INP [VIRTUAL_JUNCTIONS]). Use virtual_eligible to test whether a non-virtual node could be converted.

async openswmm_mcp.tools.nodes.set_depths_bulk(ctx: fastmcp.Context, session_id: str = 'default', depths: list[float] | None = None) → dict#

Set depths for all nodes from an array (length must equal node count).

Use case: initialize a hot-start or override an entire depth field before a step. The array is positional — index i maps to node i in storage order (see query.list_nodes for the canonical order).

async openswmm_mcp.tools.nodes.set_divider_type(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', divider_type: str = 'cutoff') → dict#

Set the divider rule type for a divider node.

divider_type: cutoff / overflow / tabular / weir or the integer code (0..3).

async openswmm_mcp.tools.nodes.set_exfil_params(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', suction: float = 0.0, ksat: float = 0.0, imd: float = 0.0) → dict#

Set Green-Ampt exfiltration params for a storage node.

Parameters:

suction – Suction head at the wetting front.
ksat – Saturated hydraulic conductivity.
imd – Initial moisture deficit.

async openswmm_mcp.tools.nodes.set_head_boundary(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', head: float = 0.0) → dict#

Apply a one-shot head boundary value at a node for the current step.

Runs against the running simulation; the value applies to the next routing step only (it is not persistent).

async openswmm_mcp.tools.nodes.set_lat_inflows_bulk(ctx: fastmcp.Context, session_id: str = 'default', inflows: list[float] | None = None) → dict#

Set lateral inflows for all nodes from an array.

Positional, length must equal node count.

async openswmm_mcp.tools.nodes.set_outfall_flap_gate(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', has_gate: bool = False) → dict#: Set whether an outfall has a flap gate.

async openswmm_mcp.tools.nodes.set_outfall_route_to(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', subcatch_index: int = -1) → dict#: Route outfall discharge to a subcatchment (-1 = none).

async openswmm_mcp.tools.nodes.set_outfall_stage(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', stage: float = 0.0) → dict#: Set the fixed stage elevation for a FIXED outfall.

async openswmm_mcp.tools.nodes.set_outfall_tidal(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', curve_index: int = 0) → dict#: Assign a tidal curve to a TIDAL outfall (hour-of-day vs stage).

async openswmm_mcp.tools.nodes.set_outfall_timeseries(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', timeseries_index: int = 0) → dict#: Assign a time series to a TIMESERIES outfall (time vs stage).

async openswmm_mcp.tools.nodes.set_outfall_type(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', outfall_type: str = 'free') → dict#

Set the outfall boundary type for an outfall node.

outfall_type: free / normal / fixed / tidal / timeseries or the integer code (0..4).

async openswmm_mcp.tools.nodes.set_quality_mass_flux(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', pollutant_index: int = 0, mass_flux: float = 0.0) → dict#

Inject a persistent pollutant mass flux at a node (mass/sec).

Runs against the running simulation; persists until cleared.

async openswmm_mcp.tools.nodes.set_storage_curve(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', curve_index: int = 0) → dict#

Assign a storage curve to a storage node.

curve_index references a curve defined via tables.add_curve (with curve_type='storage').

async openswmm_mcp.tools.nodes.set_storage_functional(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', a: float = 0.0, b: float = 0.0, c: float = 0.0) → dict#: Set functional storage params area = a * depth^b + c.

async openswmm_mcp.tools.nodes.set_storage_geometry(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', shape: str = '', p1: float = 0.0, p2: float = 0.0, p3: float = 0.0) → dict#

Set a storage node’s geometric surface-area relation.

When shape is given it is applied first — which detaches any storage curve and re-derives the internal area coefficients — then p1/p2/ p3 are supplied. Leave shape empty to redimension the node’s current shape. See get_storage_geometry for the per-shape meaning of the three dimensions.

Valid shapes here are the geometric ones: cylindrical, conical, paraboloid, pyramidal. Use set_storage_curve for tabular and set_storage_functional for functional.

The engine requires p1 > 0, p2 > 0, p3 >= 0, and p3 != 0 for paraboloid.

async openswmm_mcp.tools.nodes.set_storage_seep_rate(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', rate: float = 0.0) → dict#: Set the seepage rate for a storage node.

async openswmm_mcp.tools.nodes.set_tag(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', tag: str = '') → dict#

Set (or clear) the free-form tag string for a node.

An empty string clears the tag.

async openswmm_mcp.tools.nodes.stat_max_depth(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the peak depth recorded for a node over the simulation.

async openswmm_mcp.tools.nodes.stat_max_overflow(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the peak overflow rate for a node over the simulation.

async openswmm_mcp.tools.nodes.stat_time_flooded(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the total flooded duration (hours) for a node.

async openswmm_mcp.tools.nodes.stat_vol_flooded(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#: Return the total flooded volume for a node over the simulation.

async openswmm_mcp.tools.nodes.virtual_eligible(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '') → dict#

Dry-run check of the virtual-junction usage rules for a node.

Read-only; nothing is changed. eligible is True (rule_code 0) when the node satisfies every structural rule — exactly two attached conduits of identical cross-section, zero offsets, no lateral inflow sources, dynamic-wave routing — so converting it would succeed. Otherwise rule_code is the distinct ERR_VJ_* code identifying the violated rule (609 = not exactly two conduits, 611 = cross-section mismatch, 613 = nonzero offset, 617 = a lateral inflow source targets the node).

openswmm_mcp.tools.links#

Links tools: fine-grained accessors beyond query.get_link_info.

The aggregate query.get_link_info returns a batched property dict for any link, and editing.set_link_properties handles the basic geometry setters (length / roughness / offsets / initial_flow / max_flow / xsect_shape + geom). This module adds the fine-grained pieces those aggregates don’t cover:

Statistics (8) — peaks / durations + pump stats + hydraulic power.
Bulk arrays (4) — depths / flows / quality reads, flows writes.
Control / open-close state (5) — control_setting (continuous), target_setting (transition target), closed (binary).
Pump subtype (4) — pump_curve, pump_init_state.
Conduit detail (16) — barrels, culvert_code, loss_coeff, seep_rate, end_contractions, crest_height, discharge_coeff, flap_gate.
Quality (1) — pollutant concentration getter.

Tools accept either an integer link index or a string link ID; resolution is via session.links.get_index. Many setters are RUNNING-state-only (per the C contract); the tool layer enforces this with require_state.

async openswmm_mcp.tools.links.get_barrels(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the number of parallel barrels for a conduit.

async openswmm_mcp.tools.links.get_closed(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return whether a link is currently closed (no flow).

async openswmm_mcp.tools.links.get_control_setting(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the current continuous control setting (e.g. pump speed, orifice opening).

async openswmm_mcp.tools.links.get_control_settings_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current control settings for all links as {id, index, value} records.

async openswmm_mcp.tools.links.get_crest_height(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the crest height for a weir link.

async openswmm_mcp.tools.links.get_culvert_code(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the FHWA HDS-5 culvert inlet code (0 = not a culvert).

async openswmm_mcp.tools.links.get_depths_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current depths for all links.

async openswmm_mcp.tools.links.get_discharge_coeff(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the discharge coefficient (Cd) for a weir / orifice.

async openswmm_mcp.tools.links.get_end_contractions(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the number of end contractions on a weir (0 / 1 / 2).

async openswmm_mcp.tools.links.get_flap_gate(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return whether a conduit / orifice has a flap gate.

async openswmm_mcp.tools.links.get_flows_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current flows for all links as {id, index, value} records.

async openswmm_mcp.tools.links.get_ids_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the list of all link IDs in index order.

async openswmm_mcp.tools.links.get_loss_coeff(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the conduit’s head-loss coefficients as (inlet, outlet, avg).

async openswmm_mcp.tools.links.get_orifice_open_close_rate(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the orifice open/close rate (time, in hours, to fully operate the gate).

async openswmm_mcp.tools.links.get_outlet_expon(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the outlet rating-curve exponent (functional rating types only).

async openswmm_mcp.tools.links.get_outlet_rating_type(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#

Return the outlet rating-curve classification.

rating_type enum: 0=FUNCTIONAL_HEAD, 1=FUNCTIONAL_DEPTH, 2=TABULAR_HEAD, 3=TABULAR_DEPTH.

async openswmm_mcp.tools.links.get_pump_curve(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the pump-curve index for a pump link.

async openswmm_mcp.tools.links.get_pump_init_state(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the initial ON/OFF state of a pump (1 = on, 0 = off).

async openswmm_mcp.tools.links.get_pump_shutoff_depth(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the wet-well depth below which the pump shuts off.

async openswmm_mcp.tools.links.get_pump_startup_depth(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the wet-well depth above which the pump starts up.

async openswmm_mcp.tools.links.get_quality(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', pollutant_index: int = 0) → dict#: Return the current concentration of a pollutant in a link.

async openswmm_mcp.tools.links.get_quality_bulk(ctx: fastmcp.Context, session_id: str = 'default', pollutant_index: int = 0) → dict#: Return pollutant concentrations across all links for one pollutant.

async openswmm_mcp.tools.links.get_seep_rate(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the conduit seepage rate (depth/time).

async openswmm_mcp.tools.links.get_tag(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the link’s free-form tag string (empty string when unset).

async openswmm_mcp.tools.links.get_target_setting(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the target setting (the value the link is transitioning toward).

async openswmm_mcp.tools.links.get_target_settings_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current target settings for all links as {id, index, value} records.

async openswmm_mcp.tools.links.get_xsect(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#

Return a link’s cross-section shape plus its four geometry parameters.

shape is the XSectShape enum name; shape_code is its integer value. g1..g4 are the shape-dependent geometry values (for most closed conduits g1 is the full depth / max height).

async openswmm_mcp.tools.links.hyd_power(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#

Return the current hydraulic power dissipated in a link.

Unlike the stat_* tools, hyd_power lives directly on the Link (not under link.stats) — it’s the instantaneous value, not a cumulative statistic.

async openswmm_mcp.tools.links.set_barrels(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', barrels: int = 1) → dict#: Set the number of parallel barrels for a conduit.

async openswmm_mcp.tools.links.set_closed(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', closed: bool = False) → dict#: Close or open a link (binary on/off state).

async openswmm_mcp.tools.links.set_control_setting(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', setting: float = 0.0) → dict#

Set the control setting (typically 0..1) on a link.

Distinct from forcing.set_link_control / controls.set_link_setting only in namespace — all three wrap the same C call. Use this when working primarily through the links namespace.

async openswmm_mcp.tools.links.set_crest_height(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', crest_height: float = 0.0) → dict#: Set the weir crest height.

async openswmm_mcp.tools.links.set_culvert_code(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', culvert_code: int = 0) → dict#: Set the FHWA HDS-5 culvert inlet code.

async openswmm_mcp.tools.links.set_discharge_coeff(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', discharge_coeff: float = 0.0) → dict#: Set the discharge coefficient (Cd).

async openswmm_mcp.tools.links.set_end_contractions(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', end_contractions: int = 0) → dict#: Set the number of end contractions on a weir.

async openswmm_mcp.tools.links.set_flap_gate(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', has_flap_gate: bool = False) → dict#: Set the flap-gate flag (prevents backflow).

async openswmm_mcp.tools.links.set_flows_bulk(ctx: fastmcp.Context, session_id: str = 'default', flows: list[float] | None = None) → dict#: Set flows for all links from a positional array (length = link count).

async openswmm_mcp.tools.links.set_loss_coeff(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', inlet: float = 0.0, outlet: float = 0.0, avg: float = 0.0) → dict#: Set the conduit head-loss coefficients (entrance, exit, average).

async openswmm_mcp.tools.links.set_orifice_open_close_rate(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', open_close_rate: float = 0.0) → dict#: Set the orifice open/close rate (hours to fully operate the gate).

async openswmm_mcp.tools.links.set_outlet_expon(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', expon: float = 0.0) → dict#: Set the outlet rating-curve exponent.

async openswmm_mcp.tools.links.set_outlet_rating_type(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', rating_type: int = 0) → dict#

Set the outlet rating-curve classification (integer code 0..3).

0=FUNCTIONAL_HEAD, 1=FUNCTIONAL_DEPTH, 2=TABULAR_HEAD, 3=TABULAR_DEPTH.

async openswmm_mcp.tools.links.set_pump_curve(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', curve_index: int = 0) → dict#: Assign a pump curve to a pump link (curve type PUMP1..PUMP4).

async openswmm_mcp.tools.links.set_pump_init_state(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', init_on: bool = False) → dict#: Set the initial ON/OFF state of a pump.

async openswmm_mcp.tools.links.set_pump_shutoff_depth(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', shutoff_depth: float = 0.0) → dict#: Set the wet-well depth below which the pump shuts off.

async openswmm_mcp.tools.links.set_pump_startup_depth(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', startup_depth: float = 0.0) → dict#: Set the wet-well depth above which the pump starts up.

async openswmm_mcp.tools.links.set_seep_rate(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', seep_rate: float = 0.0) → dict#: Set the conduit seepage rate.

async openswmm_mcp.tools.links.set_tag(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', tag: str = '') → dict#: Set the link’s free-form tag string (empty string clears it).

async openswmm_mcp.tools.links.set_target_setting(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '', target: float = 0.0) → dict#: Set the gradual-transition target setting on a link.

async openswmm_mcp.tools.links.stat_max_filling(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the peak depth/full-depth ratio (0..1+) for a conduit.

async openswmm_mcp.tools.links.stat_max_flow(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the peak flow recorded for a link over the simulation.

async openswmm_mcp.tools.links.stat_max_velocity(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the peak velocity for a link.

async openswmm_mcp.tools.links.stat_pump_cycles(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the on/off cycle count for a pump link.

async openswmm_mcp.tools.links.stat_pump_on_time(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return total on-time (seconds) for a pump link.

async openswmm_mcp.tools.links.stat_pump_volume(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return total volume pumped by a pump link.

async openswmm_mcp.tools.links.stat_surcharge_time(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return total surcharge duration (hours) for a link.

async openswmm_mcp.tools.links.stat_vol_flow(ctx: fastmcp.Context, session_id: str = 'default', link_id: str | int = '') → dict#: Return the total volume conveyed through a link.

openswmm_mcp.tools.subcatchments#

Subcatchments tools: fine-grained accessors beyond query.get_subcatchment_info.

The aggregate query.get_subcatchment_info returns a batched property dict and editing.set_subcatchment_properties handles the basic geometry + roughness + outlet setters. This module adds:

Statistics (3) — precipitation / runoff peaks (via stats sub-view).
Bulk arrays (2) — runoff + quality across all subcatchments.
Current state (6) — runoff, rainfall, evap, groundwater, snow_depth, infil.
Coverage (3) — land-use coverage get/set via the coverage mapping, plus the bulk coverages() reader.
Infiltration models (8) — model getter + (Horton / Green-Ampt / Curve Number) parameter pairs (via infiltration sub-view).
Quality (3) — ponded quality get/set + per-subcatchment quality.
Initial loading (2) — [LOADINGS] initial buildup get/set.
Aquifer definitions (6) — add / id / numeric params / evap pattern (via session.aquifers).
Snowpack definitions (9) — add / count / id, the three snow-melt surfaces, and the REMOVAL row (via session.snowpacks).

async openswmm_mcp.tools.subcatchments.aquifer_add(ctx: fastmcp.Context, session_id: str = 'default', aquifer_id: str = '') → dict#

Add a new [AQUIFERS] entry with default parameters.

Returns the new aquifer’s zero-based index. Configure it with aquifer_set_param / aquifer_set_evap_pattern, then attach it to a subcatchment with set_aquifer.

async openswmm_mcp.tools.subcatchments.aquifer_get_evap_pattern(ctx: fastmcp.Context, session_id: str = 'default', aquifer_id: str | int = '') → dict#

Return an aquifer’s upper-zone evaporation pattern name (empty if none).

The trailing ETupat column of the [AQUIFERS] line — a MONTHLY [PATTERNS] name scaling the upper-zone evaporation fraction. The 12 numeric columns are reached via aquifer_get_param.

async openswmm_mcp.tools.subcatchments.aquifer_get_param(ctx: fastmcp.Context, session_id: str = 'default', aquifer_id: str | int = '', param: str | int = 'porosity') → dict#

Return an aquifer parameter value (input-file units).

aquifer_id is an aquifer name or index ([AQUIFERS] section). param is one of: porosity, wilting_point, field_capacity, conductivity, conduct_slope, tension_slope, upper_evap_frac, lower_evap_depth, lower_loss_coeff, bottom_elev, water_table_elev, upper_moisture (or the integer code 0..11).

async openswmm_mcp.tools.subcatchments.aquifer_id(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the string id of the index-th aquifer.

async openswmm_mcp.tools.subcatchments.aquifer_set_evap_pattern(ctx: fastmcp.Context, session_id: str = 'default', aquifer_id: str | int = '', pattern_id: str = '') → dict#

Set (or clear) an aquifer’s upper-zone evaporation pattern.

pattern_id is a MONTHLY [PATTERNS] name; an empty string clears it. Pre-start-only — the engine raises while the simulation is running.

async openswmm_mcp.tools.subcatchments.aquifer_set_param(ctx: fastmcp.Context, session_id: str = 'default', aquifer_id: str | int = '', param: str | int = 'porosity', value: float = 0.0) → dict#

Set an aquifer parameter value (input-file units).

aquifer_id is an aquifer name or index. param accepts the same tokens as aquifer_get_param. Flux-coefficient parameters take effect on the next step mid-run; structural / initial-condition parameters are pre-start-only and the engine raises while the simulation is running.

async openswmm_mcp.tools.subcatchments.get_aquifer(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the aquifer index assigned to a subcatchment (-1 if none).

async openswmm_mcp.tools.subcatchments.get_coverage(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', landuse_index: int = 0) → dict#: Return the land-use coverage fraction (0..1) for a (subcatch, landuse) pair.

async openswmm_mcp.tools.subcatchments.get_coverages(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#

Return every land-use coverage for a subcatchment in one call.

Bulk peer of get_coverage: coverages[i] is the coverage of land-use index i, in PERCENT (0-100) as stored in the INP [COVERAGES] section. Resolve the land-use names with quality_landuse_id.

async openswmm_mcp.tools.subcatchments.get_evap(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the current evaporation rate for a subcatchment.

async openswmm_mcp.tools.subcatchments.get_groundwater(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the current groundwater flow for a subcatchment.

async openswmm_mcp.tools.subcatchments.get_gw_node(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the node index receiving a subcatchment’s groundwater (-1 if none).

async openswmm_mcp.tools.subcatchments.get_gw_params(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#

Return the [GROUNDWATER] flow parameters for a subcatchment.

Keys: surf_elev, a1, b1, a2, b2, a3, tw, hstar. The subcatchment must have an aquifer assigned.

async openswmm_mcp.tools.subcatchments.get_ids_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the IDs of all subcatchments in storage order as {count, ids}.

async openswmm_mcp.tools.subcatchments.get_infil(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the current infiltration rate for a subcatchment.

async openswmm_mcp.tools.subcatchments.get_infil_curve_number(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#

Return the SCS Curve Number and drying time for a subcatchment.

drying_time is the third [INFILTRATION] column – days for a fully saturated soil to dry – and is what set_infil_curve_number preserves when it is not given one.

async openswmm_mcp.tools.subcatchments.get_infil_green_ampt(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return Green-Ampt params (suction, conductivity, initial_deficit).

async openswmm_mcp.tools.subcatchments.get_infil_horton(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return Horton infiltration params (f0, fmin, decay, dry_time).

async openswmm_mcp.tools.subcatchments.get_infil_model(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#

Return the infiltration model type for a subcatchment.

Model codes: 0=HORTON, 1=MOD_HORTON, 2=GREEN_AMPT, 3=MOD_GREEN_AMPT, 4=CURVE_NUMBER.

async openswmm_mcp.tools.subcatchments.get_initial_loading(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', pollutant_id: str | int = 0) → dict#

Return the [LOADINGS] initial pollutant buildup on a subcatchment.

The mass per unit area present at simulation start (0.0 when unset), which overrides the DRY_DAYS-derived buildup. pollutant_id is a pollutant name or index.

async openswmm_mcp.tools.subcatchments.get_ponded_quality(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', pollutant_index: int = 0) → dict#: Return the ponded pollutant mass on a subcatchment surface.

async openswmm_mcp.tools.subcatchments.get_quality(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', pollutant_index: int = 0) → dict#: Return the runoff pollutant concentration for a subcatchment.

async openswmm_mcp.tools.subcatchments.get_quality_bulk(ctx: fastmcp.Context, session_id: str = 'default', pollutant_index: int = 0) → dict#: Return pollutant concentrations across all subcatchments for one pollutant.

async openswmm_mcp.tools.subcatchments.get_rainfall(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the current rainfall rate for a subcatchment.

async openswmm_mcp.tools.subcatchments.get_runoff(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the current runoff rate for a subcatchment.

async openswmm_mcp.tools.subcatchments.get_runoff_bulk(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return current runoff rates for all subcatchments.

async openswmm_mcp.tools.subcatchments.get_snow_depth(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the current snow depth on a subcatchment.

async openswmm_mcp.tools.subcatchments.get_tag(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#

Return the free-form tag string for a subcatchment (empty if untagged).

Tags come from the INP [TAGS] section and are keyed by index.

async openswmm_mcp.tools.subcatchments.get_zero_imperv_pct(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#

Return the [SUBAREAS] PctZero value for a subcatchment.

The percentage (0-100) of the impervious area that has no depression storage.

async openswmm_mcp.tools.subcatchments.set_aquifer(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', aquifer: str | int = -1) → dict#

Assign (or detach) the aquifer for a subcatchment.

Parameters:: aquifer – Aquifer name or index. Pass -1 (or an empty string) to detach the aquifer (no groundwater).

async openswmm_mcp.tools.subcatchments.set_coverage(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', landuse_index: int = 0, fraction: float = 0.0) → dict#: Set the land-use coverage fraction (0..1) for a (subcatch, landuse) pair.

async openswmm_mcp.tools.subcatchments.set_gw_node(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', node: str | int = -1) → dict#

Set (or detach) the node receiving a subcatchment’s groundwater flow.

Parameters:: node – Node name or index. Pass -1 (or an empty string) to detach.

async openswmm_mcp.tools.subcatchments.set_gw_params(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', surf_elev: float = 0.0, a1: float = 0.0, b1: float = 0.0, a2: float = 0.0, b2: float = 0.0, a3: float = 0.0, tw: float = 0.0, hstar: float = 0.0) → dict#

Set the [GROUNDWATER] flow parameters for a subcatchment.

Token order matches the INP [GROUNDWATER] section. The subcatchment must have an aquifer assigned.

async openswmm_mcp.tools.subcatchments.set_gw_state(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', theta: float = -1.0, lower_depth: float = -1.0) → dict#

Inject the groundwater state on a subcatchment (running only).

Overwrites the live upper-zone moisture and/or saturated-zone depth so a caller can warm-start or perturb groundwater mid-run. Pass a negative value to leave that field unchanged.

Parameters:

subcatch_id – Subcatchment name or index.
theta – Upper-zone moisture content (0..porosity); negative leaves it as-is.
lower_depth – Saturated-zone depth above the aquifer bottom in project length units; negative leaves it as-is.

async openswmm_mcp.tools.subcatchments.set_infil_curve_number(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', curve_number: float = 0.0, drying_time: float | None = None) → dict#

Set the SCS Curve Number for a subcatchment.

The engine writes both [INFILTRATION] columns in one call. Leave drying_time unset to keep the subcatchment’s current value and change only the curve number.

async openswmm_mcp.tools.subcatchments.set_infil_green_ampt(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', suction: float = 0.0, conductivity: float = 0.0, initial_deficit: float = 0.0) → dict#: Set Green-Ampt infiltration params for a subcatchment.

async openswmm_mcp.tools.subcatchments.set_infil_horton(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', f0: float = 0.0, fmin: float = 0.0, decay: float = 0.0, dry_time: float = 0.0) → dict#: Set Horton infiltration params for a subcatchment.

async openswmm_mcp.tools.subcatchments.set_infil_model(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', model: int = 0) → dict#

Switch the active infiltration model for a subcatchment.

Parameters:: model – InfilModel integer code (e.g. 0=HORTON, per the engine enum). Per-model parameter sub-arrays are preserved.

async openswmm_mcp.tools.subcatchments.set_initial_loading(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', pollutant_id: str | int = 0, initial_loading: float = 0.0) → dict#

Set the [LOADINGS] initial pollutant buildup on a subcatchment.

initial_loading is the buildup mass per unit area present at simulation start. pollutant_id is a pollutant name or index.

async openswmm_mcp.tools.subcatchments.set_outlet_subcatchment(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', outlet_subcatch_id: str | int = '') → dict#

Route a subcatchment’s runoff to another subcatchment.

outlet_subcatch_id is the receiving subcatchment’s name or index.

async openswmm_mcp.tools.subcatchments.set_ponded_quality(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', pollutant_index: int = 0, ponded_mass: float = 0.0) → dict#: Set the ponded pollutant mass on a subcatchment surface.

async openswmm_mcp.tools.subcatchments.set_snow_state(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', surface: int = 2, swe: float = -1.0, free_water: float = -1.0, ati: float = -1000.0, cold_content: float = -1.0) → dict#

Inject the snow-pack state on one snow surface (running only).

Overwrites the live snow-pack state on a single snow subarea so a caller can warm-start or perturb the snowpack mid-run. Pass the documented sentinel (negative for depths, -1000 for ATI) to leave a field unchanged.

Parameters:

subcatch_id – Subcatchment name or index.
surface – Snow subarea: 0 plowable, 1 impervious, 2 pervious (default).
swe – Snow water equivalent in project depth units; negative leaves as-is.
free_water – Free water in project depth units; negative leaves as-is.
ati – Antecedent temperature index (deg F US, deg C SI); -1000 leaves as-is.
cold_content – Cold content in project depth units of melt equivalent; negative leaves as-is.

async openswmm_mcp.tools.subcatchments.set_tag(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', tag: str = '') → dict#

Set (or clear) the free-form tag string for a subcatchment.

An empty string clears the tag.

async openswmm_mcp.tools.subcatchments.set_zero_imperv_pct(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', pct: float = 0.0) → dict#

Set the [SUBAREAS] PctZero value for a subcatchment.

pct is the percentage (0-100) of the impervious area having no depression storage.

async openswmm_mcp.tools.subcatchments.snowpack_add(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str = '') → dict#

Add a new [SNOWPACKS] definition with zeroed parameters.

Returns the new snowpack’s zero-based index. Configure the three snow-melt surfaces with snowpack_set_surface and the redistribution row with snowpack_set_removal.

async openswmm_mcp.tools.subcatchments.snowpack_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of [SNOWPACKS] definitions in the model.

async openswmm_mcp.tools.subcatchments.snowpack_get_removal(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str | int = '') → dict#: Read a snowpack’s REMOVAL row (snow redistribution fractions).

async openswmm_mcp.tools.subcatchments.snowpack_get_removal_subcatch(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str | int = '') → dict#: Return the destination subcatchment for a snowpack’s fsubcatch removal fraction (empty if none).

async openswmm_mcp.tools.subcatchments.snowpack_get_surface(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str | int = '', surface: str | int = 'pervious') → dict#

Read one snow-melt surface of a snowpack definition.

surface is plowable, impervious or pervious (or the codes 0, 1, 2). Returns the seven [SNOWPACKS] values — see snowpack_set_surface for their meaning.

async openswmm_mcp.tools.subcatchments.snowpack_id(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the string id of the index-th snowpack.

async openswmm_mcp.tools.subcatchments.snowpack_set_removal(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str | int = '', dsnow: float = 0.0, fout: float = 0.0, fimp: float = 0.0, fperv: float = 0.0, fimelt: float = 0.0, fsubcatch: float = 0.0) → dict#

Set a snowpack’s REMOVAL row (pre-start-only).

Parameters:

dsnow – Snow depth above which removal begins (in or mm).
fout – Fractions of the removed snow routed out of the watershed, to the impervious area, to the pervious area, converted to immediate melt, and transferred to another subcatchment. Name the destination subcatchment with snowpack_set_removal_subcatch.
fimp – Fractions of the removed snow routed out of the watershed, to the impervious area, to the pervious area, converted to immediate melt, and transferred to another subcatchment. Name the destination subcatchment with snowpack_set_removal_subcatch.
fperv – Fractions of the removed snow routed out of the watershed, to the impervious area, to the pervious area, converted to immediate melt, and transferred to another subcatchment. Name the destination subcatchment with snowpack_set_removal_subcatch.
fimelt – Fractions of the removed snow routed out of the watershed, to the impervious area, to the pervious area, converted to immediate melt, and transferred to another subcatchment. Name the destination subcatchment with snowpack_set_removal_subcatch.
fsubcatch – Fractions of the removed snow routed out of the watershed, to the impervious area, to the pervious area, converted to immediate melt, and transferred to another subcatchment. Name the destination subcatchment with snowpack_set_removal_subcatch.

async openswmm_mcp.tools.subcatchments.snowpack_set_removal_subcatch(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str | int = '', subcatch_id: str = '') → dict#

Set (or clear) the destination subcatchment for a snowpack’s fsubcatch removal fraction.

An empty subcatch_id clears it. Pre-start-only.

async openswmm_mcp.tools.subcatchments.snowpack_set_surface(ctx: fastmcp.Context, session_id: str = 'default', snowpack_id: str | int = '', surface: str | int = 'pervious', cmin: float = 0.0, cmax: float = 0.0, tbase: float = 0.0, fwfrac: float = 0.0, sd0: float = 0.0, fw0: float = 0.0, last: float = 0.0) → dict#

Set one snow-melt surface of a snowpack definition (pre-start-only).

Parameters:

surface – plowable, impervious or pervious (or the codes 0, 1, 2).
cmin – Minimum / maximum melt coefficient (in or mm per hr per degree).
cmax – Minimum / maximum melt coefficient (in or mm per hr per degree).
tbase – Snow-melt base temperature (deg F or C).
fwfrac – Free-water capacity as a fraction of snow depth.
sd0 – Initial snow depth (in or mm water equivalent).
fw0 – Initial free water (in or mm).
last – PLOWABLE: the fraction of the impervious area that is plowable. IMPERVIOUS / PERVIOUS: the snow depth above which there is 100% cover (in or mm).

async openswmm_mcp.tools.subcatchments.stat_max_runoff(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return the peak runoff rate for a subcatchment.

async openswmm_mcp.tools.subcatchments.stat_precip(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return total precipitation volume for a subcatchment.

async openswmm_mcp.tools.subcatchments.stat_runoff_vol(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '') → dict#: Return total runoff volume for a subcatchment.

openswmm_mcp.tools.inflows#

Inflows tools: external inflows, dry-weather flow, RDII, and unit hydrographs.

Wraps openswmm.engine.Inflows for design-time configuration of boundary inflows. Runtime overrides (rainfall, head BCs, lateral inflow per step) live in forcing.py; this module is the persistent .inp-section counterpart.

Domain covered (Python Inflows surface):

[INFLOWS] — add_external(), ext_inflow_count()
[DWF] — add_dwf(), dwf_count()
[RDII] — add_rdii(), get_rdii(), rdii_count()
[HYDROGRAPHS] — add_hydrograph(), get_hydrograph(), hydrograph_count(), add_hydrograph_gage(), get_hydrograph_gage(), hydrograph_gage_count(), hydrograph_group_count(), list_hydrograph_groups()
[RDII_DECAY] — add_rdii_decay(), get_rdii_decay(), rdii_decay_count()

The Python Inflows constructor accepts either a Solver (any non-closed lifecycle state) or a ModelBuilder (building state). The _get_inflows_accessor() helper picks the right binding.

Node references: external inflow / DWF / RDII tools accept the string node ID and resolve to integer index via session.nodes.get_index before calling the engine (which takes int only). Pattern / time series / hydrograph names are passed through unchanged — verifying that those upstream tables exist is the caller’s responsibility (a future hardening pass could cross-check via tables.get_index before forwarding).

async openswmm_mcp.tools.inflows.add_dwf(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', constituent: str = 'FLOW', avg_value: float = 0.0, monthly_pattern: str = '', daily_pattern: str = '', hourly_pattern: str = '', weekend_pattern: str = '') → dict#

Add a dry-weather flow to a node.

constituent is "FLOW" or a pollutant ID. avg_value is the constant baseline value. The four pattern arguments are pattern IDs (empty string = unused); use tables.pattern_add to create them.

Pattern coupling: this tool does not verify that the named patterns exist before forwarding to the engine. A dangling reference will be surfaced by the engine at lookup time, not here.

async openswmm_mcp.tools.inflows.add_external(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', constituent: str = 'FLOW', ts_name: str = '', inflow_type: str = 'FLOW', m_factor: float = 1.0, s_factor: float = 1.0, baseline: float = 0.0, pattern: str = '') → dict#

Add an external inflow to a node.

constituent is either "FLOW" or a pollutant ID. inflow_type is one of "FLOW", "CONCEN", "MASS". ts_name references an existing [TIMESERIES] entry (empty string for none).

async openswmm_mcp.tools.inflows.add_hydrograph(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', month: str | int = 'all', response: str | int = 'short', r: float = 0.0, t: float = 0.0, k: float = 1.0, dmax: float = 0.0, drecov: float = 0.0, dinit: float = 0.0) → dict#

Add a unit-hydrograph parameter line.

month is "all"/-1 (the default), "jan"..``”dec”, or ``0..11. response is "short"/"medium"/"long" or 0..2.

Parameters:

r – Fraction of rainfall that becomes RDII (0..1).
t – Time to peak (hours).
k – Ratio of base time to peak time (must be >= 1).
dmax – Maximum initial-abstraction depth (project depth units).
drecov – Linear-model IA recovery rate. Ignored when an exponential-decay row exists for the same (uh_name, response) pair (see add_rdii_decay()).
dinit – Initial IA already used.

async openswmm_mcp.tools.inflows.add_hydrograph_gage(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', gage_name: str = '') → dict#

Assign a rain gage to a unit-hydrograph group.

The gage drives the RDII calculation for every node that references this hydrograph via add_rdii().

async openswmm_mcp.tools.inflows.add_rdii(ctx: fastmcp.Context, session_id: str = 'default', node_id: str | int = '', uh_name: str = '', area: float = 0.0) → dict#

Assign an RDII inflow to a node.

uh_name references a unit hydrograph group created via add_hydrograph(). area is the contributing sewershed area (project area units).

async openswmm_mcp.tools.inflows.add_rdii_decay(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', response: str | int = 'short', k_dep: float = 0.0, k_0: float = 0.0, k_T: float = 0.0, T_ref: float = 10.0, theta_rec: float = 0.0, T_freeze: float = 0.0) → dict#

Add an exponential IA-decay row for a (uh_name, response) pair.

Replaces the legacy linear drecov rate from add_hydrograph() with a physically-based recovery model:

depletion: dIA/dt = -k_dep * rainfall recovery: dIA/dt = +k_0 + k_T * exp(theta_rec * (T - T_ref))

Recovery is suppressed when T <= T_freeze. The hydrograph row for (uh_name, response) must already exist.

async openswmm_mcp.tools.inflows.clear_hydrograph_group_months(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '') → dict#: Clear the month-specific rows of a UH group, keeping the ALL-months row.

async openswmm_mcp.tools.inflows.dwf_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of dry-weather-flow rows in the model.

async openswmm_mcp.tools.inflows.ext_inflow_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of external inflow rows in the model.

async openswmm_mcp.tools.inflows.get_dwf(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#

Read back the I{entry_index}-th dry-weather-flow row as a dict.

Returns the persisted [DWF] row: node, constituent, average value, and the four pattern IDs (monthly / daily / hourly / weekend).

async openswmm_mcp.tools.inflows.get_external(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#

Read back the I{entry_index}-th external inflow row as a dict.

Returns the persisted [INFLOWS] row: node, constituent, time-series name, inflow type, scale (m_factor), unit conversion (s_factor), baseline, and baseline pattern.

async openswmm_mcp.tools.inflows.get_hydrograph(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Read back the I{entry_index}-th hydrograph row as a dict.

async openswmm_mcp.tools.inflows.get_hydrograph_gage(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Read back the I{entry_index}-th UH-to-gage assignment as (uh_name, gage_name).

async openswmm_mcp.tools.inflows.get_rdii(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Read back the I{entry_index}-th RDII assignment as (node_idx, uh_name, area).

async openswmm_mcp.tools.inflows.get_rdii_decay(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Read back the I{entry_index}-th exponential-decay row as a dict.

async openswmm_mcp.tools.inflows.hydrograph_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of hydrograph parameter rows in the model.

async openswmm_mcp.tools.inflows.hydrograph_gage_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of UH-to-gage assignments in the model.

async openswmm_mcp.tools.inflows.hydrograph_group_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the number of unique unit-hydrograph group names.

A UH “group” is identified by name; the engine stores one row per (group, month, response). This count is the number of distinct group names across parameter entries and gage assignments — the figure a GUI Object Browser needs for the Unit Hydrographs section.

async openswmm_mcp.tools.inflows.list_hydrograph_groups(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the unit-hydrograph groups as a list of {index, name} dicts.

Groups are enumerated in first-occurrence order across the parameter entry list (matches the order in which they appear in the [HYDROGRAPHS] section of the input file). Convenience wrapper that batches hydrograph_group_count + N x get_hydrograph_group_id so an LLM (or GUI Object Browser) can populate the Unit Hydrographs node in a single call.

async openswmm_mcp.tools.inflows.rdii_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of RDII inflow rows in the model.

async openswmm_mcp.tools.inflows.rdii_decay_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of exponential IA-decay rows in the model.

async openswmm_mcp.tools.inflows.remove_dwf(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Remove the I{entry_index}-th dry-weather-flow row.

async openswmm_mcp.tools.inflows.remove_external(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Remove the I{entry_index}-th external inflow row.

async openswmm_mcp.tools.inflows.remove_hydrograph_entry(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', month: str | int = 'all', response: str | int = 'short') → dict#: Remove a single (uh_name, month, response) hydrograph parameter row.

async openswmm_mcp.tools.inflows.remove_hydrograph_group(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '') → dict#: Remove an entire unit-hydrograph group (all rows + its gage assignment).

async openswmm_mcp.tools.inflows.remove_rdii(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0) → dict#: Remove the I{entry_index}-th RDII assignment.

async openswmm_mcp.tools.inflows.remove_rdii_decay(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', response: str | int = 'short') → dict#: Remove the exponential IA-decay row for a (uh_name, response) pair.

async openswmm_mcp.tools.inflows.rename_hydrograph_group(ctx: fastmcp.Context, session_id: str = 'default', group_index: int = 0, new_id: str = '') → dict#

Rename the I{group_index}-th unit-hydrograph group.

group_index is the position from list_hydrograph_groups().

async openswmm_mcp.tools.inflows.set_dwf_baseline(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0, avg_value: float = 0.0) → dict#: Set the average (baseline) value of the I{entry_index}-th DWF row.

async openswmm_mcp.tools.inflows.set_external_baseline(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0, baseline: float = 0.0) → dict#: Set the baseline (constant) value of the I{entry_index}-th external inflow.

async openswmm_mcp.tools.inflows.set_external_scale(ctx: fastmcp.Context, session_id: str = 'default', entry_index: int = 0, scale: float = 1.0) → dict#

Set the time-series scale factor (s_factor) of the I{entry_index}-th external inflow.

Note: this sets s_factor (the engine’s only runtime “scale” setter), not the m_factor multiplier — m_factor is set only at add_external() time.

async openswmm_mcp.tools.inflows.set_hydrograph_gage(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', gage_name: str = '') → dict#

Set (replace) the rain gage assigned to an existing UH group.

Unlike add_hydrograph_gage() (which appends a new assignment row), this updates the gage of a group that already has one.

async openswmm_mcp.tools.inflows.set_hydrograph_ia(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', month: str | int = 'all', response: str | int = 'short', dmax: float = 0.0, drecov: float = 0.0, dinit: float = 0.0) → dict#

Update the initial-abstraction parameters of an existing UH row.

Edits dmax/drecov/dinit in place, leaving R/T/K untouched. The (uh_name, month, response) row must already exist. drecov is ignored at runtime when an exponential-decay row exists for the same (uh_name, response) pair (see add_rdii_decay()).

async openswmm_mcp.tools.inflows.set_hydrograph_rtk(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', month: str | int = 'all', response: str | int = 'short', r: float = 0.0, t: float = 0.0, k: float = 1.0) → dict#

Update the R/T/K parameters of an existing (uh_name, month, response) row.

Unlike add_hydrograph(), this edits the row in place and leaves its IA parameters (dmax/drecov/dinit) untouched. The row must already exist. month/response accept the same tokens as add_hydrograph().

async openswmm_mcp.tools.inflows.set_rdii_decay(ctx: fastmcp.Context, session_id: str = 'default', uh_name: str = '', response: str | int = 'short', k_dep: float = 0.0, k_0: float = 0.0, k_T: float = 0.0, T_ref: float = 10.0, theta_rec: float = 0.0, T_freeze: float = 0.0) → dict#

Update an existing exponential IA-decay row in place.

Same parameter meaning as add_rdii_decay(); the (uh_name, response) decay row must already exist.

openswmm_mcp.tools.pollutants#

Pollutants tools: fine-grained accessors beyond query.get_pollutant_info.

The Python Pollutants accessor exposes 23 methods covering definition (decay, concentrations in rainfall / groundwater / RDII / initial), metadata (units, molecular weight, snow-only flag, co-pollutant), and runtime quality injection at nodes / links.

This module wraps the surface as 22 MCP tools. The aggregate query.get_pollutant_info covers a read-only snapshot; this module adds the individual setters for design-time configuration plus the runtime injection setters (set_node_quality / set_link_quality) gated on the running state.

async openswmm_mcp.tools.pollutants.add(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str = '', units: str = 'mg_per_l') → dict#

Add a new pollutant to the model (BUILDING state).

units: mg_per_l (0), ug_per_l (1), or count_per_l (2).

async openswmm_mcp.tools.pollutants.count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of pollutants defined in the model.

async openswmm_mcp.tools.pollutants.get_co_pollutant(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#

Return the co-pollutant index assigned to this pollutant (-1 = none).

v1 surfaces co-pollutant as an optional (Pollutant, fraction) tuple; we project that down to the legacy (index, fraction) shape so the JSON contract is preserved (with fraction exposed as a new field).

async openswmm_mcp.tools.pollutants.get_dwf_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the concentration of this pollutant in dry-weather flow.

async openswmm_mcp.tools.pollutants.get_gw_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the concentration of this pollutant in groundwater.

async openswmm_mcp.tools.pollutants.get_init_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the initial concentration throughout the system.

async openswmm_mcp.tools.pollutants.get_kdecay(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the first-order decay coefficient (1/day) for a pollutant.

async openswmm_mcp.tools.pollutants.get_mwt(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the molecular weight of a pollutant (g/mol).

async openswmm_mcp.tools.pollutants.get_rain_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the concentration of this pollutant in rainfall.

async openswmm_mcp.tools.pollutants.get_rdii_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the concentration of this pollutant in RDII.

async openswmm_mcp.tools.pollutants.get_snow_only(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the snow-only flag for a pollutant (True = transported only in snow).

async openswmm_mcp.tools.pollutants.get_units(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '') → dict#: Return the concentration units for a pollutant (mg/L / ug/L / #/L).

async openswmm_mcp.tools.pollutants.set_co_pollutant(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', co_pollutant_index: int = -1, fraction: float = 1.0) → dict#

Assign a co-pollutant (set co_pollutant_index to -1 to clear).

v1 requires a fraction alongside the co-pollutant. Defaults to 1.0 so callers that previously only supplied an index see the same effective behaviour.

async openswmm_mcp.tools.pollutants.set_dwf_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', dwf_conc: float = 0.0) → dict#: Set the dry-weather-flow concentration.

async openswmm_mcp.tools.pollutants.set_gw_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', gw_conc: float = 0.0) → dict#: Set the groundwater concentration.

async openswmm_mcp.tools.pollutants.set_init_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', init_conc: float = 0.0) → dict#: Set the initial system-wide concentration.

async openswmm_mcp.tools.pollutants.set_kdecay(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', kdecay: float = 0.0) → dict#: Set the first-order decay coefficient (1/day).

async openswmm_mcp.tools.pollutants.set_link_quality(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', pollutant_id: str | int = '', concentration: float = 0.0) → dict#: Override a link’s pollutant concentration mid-simulation.

async openswmm_mcp.tools.pollutants.set_mwt(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', molecular_weight: float = 0.0) → dict#: Set the molecular weight (g/mol).

async openswmm_mcp.tools.pollutants.set_node_quality(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', pollutant_id: str | int = '', concentration: float = 0.0) → dict#

Override a node’s pollutant concentration mid-simulation.

Runs against the running simulation. Pass the node_id (string) and the pollutant_id (string or int index); the value is the override concentration in the pollutant’s units.

async openswmm_mcp.tools.pollutants.set_rain_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', rain_conc: float = 0.0) → dict#: Set the rainfall concentration.

async openswmm_mcp.tools.pollutants.set_rdii_conc(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', rdii_conc: float = 0.0) → dict#: Set the RDII concentration.

async openswmm_mcp.tools.pollutants.set_snow_only(ctx: fastmcp.Context, session_id: str = 'default', pollutant_id: str | int = '', snow_only: bool = False) → dict#: Set the snow-only flag for a pollutant.

openswmm_mcp.tools.quality#

Quality tools: landuse / buildup / washoff / treatment.

Complements spatial_quality.set_treatment (already in place) with the remaining openswmm.engine.Quality surface:

Landuse identity + sweep parameters.
Buildup function setters / getters (per landuse + pollutant).
Washoff function setters / getters.
Treatment read-back + clear (set is already in spatial_quality).

The Python Quality accessor takes integer indices for landuse, pollutant, and node. Tools accept either string ids (resolved via the corresponding accessor) or integer indices for pollutant / node, while landuse uses landuse_id strings.

async openswmm_mcp.tools.quality.buildup_get(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '', pollutant_id: str | int = '') → dict#: Return the buildup function parameters for a (landuse, pollutant) pair.

async openswmm_mcp.tools.quality.buildup_set(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '', pollutant_id: str | int = '', function: str = 'none', c1: float = 0.0, c2: float = 0.0, c3: float = 0.0, normalizer: str = 'per_area') → dict#

Set the buildup function for a (landuse, pollutant) pair.

function: none / power / exponential / saturation / external. normalizer: per_area / per_curb.

async openswmm_mcp.tools.quality.get_sweep_interval(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '') → dict#: Return the days-between-street-sweeps for a landuse.

async openswmm_mcp.tools.quality.get_sweep_removal(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '') → dict#: Return the sweep removal fraction (0..1) for a landuse.

async openswmm_mcp.tools.quality.landuse_add(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str = '') → dict#: Add a new landuse to the model (BUILDING state).

async openswmm_mcp.tools.quality.landuse_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of landuses defined in the model.

async openswmm_mcp.tools.quality.landuse_id(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the string id of the I{index}-th landuse.

async openswmm_mcp.tools.quality.landuse_index(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str = '') → dict#: Return the integer index for a landuse string id (-1 if not found).

async openswmm_mcp.tools.quality.set_sweep_interval(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '', days: float = 0.0) → dict#: Set the days-between-street-sweeps for a landuse.

async openswmm_mcp.tools.quality.set_sweep_removal(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '', fraction: float = 0.0) → dict#: Set the sweep removal fraction for a landuse (must be in [0, 1]).

async openswmm_mcp.tools.quality.treatment_clear(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', pollutant_id: str | int = '') → dict#: Remove the treatment expression for a (node, pollutant) pair.

async openswmm_mcp.tools.quality.treatment_get(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', pollutant_id: str | int = '') → dict#: Return the treatment expression text for a (node, pollutant) pair.

async openswmm_mcp.tools.quality.treatment_validate_expression(ctx: fastmcp.Context, session_id: str = 'default', expression: str = '') → dict#

Check a treatment expression parses, without writing it to the model.

Nothing in the engine is modified. Call this before spatial_set_treatment so a malformed expression is caught here rather than surfacing much later as an opaque run-time error.

Returns valid; when False, message is the engine’s diagnostic and column is the 0-based character offset in expression where the parse failed (-1 when the failure is not attributable to a position).

async openswmm_mcp.tools.quality.washoff_get(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '', pollutant_id: str | int = '') → dict#: Return the washoff function parameters for a (landuse, pollutant) pair.

async openswmm_mcp.tools.quality.washoff_set(ctx: fastmcp.Context, session_id: str = 'default', landuse_id: str | int = '', pollutant_id: str | int = '', function: str = 'exponential', c1: float = 0.0, c2: float = 0.0, sweep_efficiency: float = 0.0, bmp_efficiency: float = 0.0) → dict#

Set the washoff function for a (landuse, pollutant) pair.

function: exponential / rating_curve / event_mean_conc.

openswmm_mcp.tools.tables#

Tables tools: curves, time series, and patterns.

Wraps the openswmm.engine.Tables accessor with MCP tools. Tables works against either a Solver (any non-closed lifecycle state) or a ModelBuilder (the building state); the helpers below pick the right binding automatically.

C-API state contract (per openswmm_tables.h):

timeseries_add / curve_add / pattern_add require SWMM_STATE_BUILDING. Creation tools therefore require the session to be in the building state with an attached ModelBuilder.
add_point / get_point / get_point_count / clear / lookup carry no state annotation in the header and work in any state where the engine handle is alive.
Patterns: pattern_count is read-only and works anywhere; pattern_add and pattern_set_factors are creation/mutation and require building.

async openswmm_mcp.tools.tables.add_curve(ctx: fastmcp.Context, session_id: str = 'default', curve_id: str = '', curve_type: str = 'storage', x_values: list[float] | None = None, y_values: list[float] | None = None) → dict#

Create a curve and populate it with (x, y) points.

curve_type is a string (storage, diversion, tidal, rating, control, shape, pump1..``pump4``, weir) or the engine integer code directly. Requires the building state.

async openswmm_mcp.tools.tables.add_point(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '', x: float = 0.0, y: float = 0.0) → dict#: Append a single (x, y) data point to an existing table.

async openswmm_mcp.tools.tables.add_timeseries(ctx: fastmcp.Context, session_id: str = 'default', ts_id: str = '', times: list[float] | None = None, values: list[float] | None = None) → dict#

Create a time series and populate it with (time, value) points.

Requires the session to be in the building state. Points are added in input order; the engine does not sort them.

async openswmm_mcp.tools.tables.clear_points(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '') → dict#: Remove all data points from a table (the table itself remains).

async openswmm_mcp.tools.tables.count(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the number of curves and time series in the model.

The count combines both since the engine stores them in a single table namespace. Patterns are counted separately; see pattern_count.

async openswmm_mcp.tools.tables.get_id(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the string ID of a table by zero-based index.

async openswmm_mcp.tools.tables.get_index(ctx: fastmcp.Context, session_id: str = 'default', table_id: str = '') → dict#

Return the zero-based index of a table by string ID.

Returns -1 if no table with that ID exists.

async openswmm_mcp.tools.tables.get_point(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '', point_index: int = 0) → dict#: Read a single (x, y) data point from a table by point index.

async openswmm_mcp.tools.tables.get_point_count(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '') → dict#: Return the number of data points in a table.

async openswmm_mcp.tools.tables.get_points(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '') → dict#

Return all data points in a table as a list of [x, y] pairs.

Reads table.points (a NumPy array) in a single C call and projects each row to [x, y] floats for the JSON wire format.

async openswmm_mcp.tools.tables.get_type(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '') → dict#

Return the type of a table (curve kind or time series).

Surfaces Tables.get_type — a TableType enum identifying the table (e.g. STORAGE, RATING, PUMP1 for curves, or the time-series kind). table_id is a string ID or integer index. Reports both the enum type name and its integer type_code.

async openswmm_mcp.tools.tables.lookup(ctx: fastmcp.Context, session_id: str = 'default', table_id: str | int = '', x: float = 0.0) → dict#

Interpolate a Y value from a table at the given X.

Uses the engine’s cursor-optimized lookup; values outside the table’s X range clamp to the nearest endpoint.

async openswmm_mcp.tools.tables.pattern_add(ctx: fastmcp.Context, session_id: str = 'default', pattern_id: str = '', pattern_type: str = 'monthly', factors: list[float] | None = None) → dict#

Create a time pattern and (optionally) seed its multiplier factors.

pattern_type accepts a string (monthly, daily, hourly, weekend) or the integer engine code. When factors is supplied, it is applied via pattern.set_factors immediately after creation. Expected factor counts: 12 for monthly, 7 for daily, 24 for hourly / weekend.

async openswmm_mcp.tools.tables.pattern_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of time patterns in the model.

async openswmm_mcp.tools.tables.pattern_remove(ctx: fastmcp.Context, session_id: str = 'default', pattern_id: str | int = '') → dict#

Remove a time pattern by string ID or integer index (BUILDING state).

Mutation; requires the session to be in the building state (mirrors pattern_add).

async openswmm_mcp.tools.tables.pattern_set_factors(ctx: fastmcp.Context, session_id: str = 'default', pattern_index: int = 0, pattern_type: str = 'monthly', factors: list[float] | None = None) → dict#

Replace the multiplier factors of an existing time pattern.

Pattern length is determined by pattern_type; supplying a factor count that does not match will raise an engine error. The pattern_type argument was added in v1 — it tells the engine which block (monthly / daily / hourly / weekend) the factors apply to.

openswmm_mcp.tools.infrastructure#

Infrastructure tools: transects, streets, inlets, LIDs, and LID usage.

Wraps openswmm.engine.Infrastructure for design-time configuration of the [TRANSECTS], [STREETS], [INLETS], [LID_CONTROLS], and [LID_USAGE] INP sections.

Previously the MCP server only exposed spatial_quality.add_lid (which is actually a lid_usage_add — placing an LID instance on a subcatchment). The full Infrastructure surface — defining LIDs, transects, streets, and inlets, plus per-layer LID parameter setters — was unreachable. This module closes that gap.

Module coverage (17 of 17 Python Infrastructure runtime methods):

Transects: transect_count, add_transect, set_transect_roughness, add_transect_station.
Streets: street_count, add_street, set_street_params.
Inlets: inlet_count, add_inlet, set_inlet_params.
LID controls: lid_count, add_lid, and per-layer setters/getters for all six layers — {set,get}_lid_surface, {set,get}_lid_soil, {set,get}_lid_storage, {set,get}_lid_drain, {set,get}_lid_pavement, {set,get}_lid_drainmat.
LID usage: add_lid_usage (accepts subcatch_id string).

State handling: the helper accepts any non-closed state. For building sessions it constructs Infrastructure(ModelBuilder) plus a Subcatchments accessor for ID resolution on add_lid_usage; otherwise the cached backend accessors are returned.

Coexistence with spatial_quality.add_lid: that tool is left in place and continues to wrap lid_usage_add with positional subcatch lookup. The new infrastructure.add_lid is the LID-control definition tool (distinct semantic — defines what an LID is, not where it’s placed).

async openswmm_mcp.tools.infrastructure.add_inlet(ctx: fastmcp.Context, session_id: str = 'default', inlet_id: str = '', inlet_type: str = '') → dict#

Create a new inlet. Returns the assigned index.

inlet_type is a string identifying the inlet geometry family (e.g. "GRATE", "CURB", "SLOTTED", "CUSTOM" — consult the engine docs for the exact set).

async openswmm_mcp.tools.infrastructure.add_lid(ctx: fastmcp.Context, session_id: str = 'default', lid_id: str = '', lid_type: str = 'bio_cell') → dict#

Define a new LID control (not a usage).

lid_type accepts a string (bio_cell, rain_garden, green_roof, infil_trench, perm_pavement, rain_barrel, rooftop_disconn, vegetative_swale) or the integer code (0..7).

Distinct from spatial_quality.add_lid which is actually lid_usage_add — placing an LID instance on a subcatchment. This tool defines what an LID is; the layer setters (set_lid_surface(), set_lid_soil(), set_lid_storage(), set_lid_drain()) configure its hydraulic behaviour. Then use add_lid_usage() to attach instances to subcatchments.

async openswmm_mcp.tools.infrastructure.add_lid_usage(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str | int = '', lid_index: int = 0, number: int = 1, area: float = 0.0, width: float = 0.0, init_sat: float = 0.0, from_imperv: float = 1.0) → dict#

Attach number instances of an LID control to a subcatchment.

Parameters:

subcatch_id – Target subcatchment (string ID or integer index).
lid_index – Index of the LID control to deploy (from add_lid()).
number – Count of identical units to place.
area – Surface area and top width of each unit.
width – Surface area and top width of each unit.
init_sat – Initial saturation fraction in [0.0, 1.0].
from_imperv – Fraction of the subcatchment’s impervious area routed to the LID (also [0.0, 1.0]).
Validation (area must be positive; init_sat and from_imperv)
[0.0 (must lie in)
1. (1.0]; number must be >=)

async openswmm_mcp.tools.infrastructure.add_street(ctx: fastmcp.Context, session_id: str = 'default', street_id: str = '') → dict#: Create a new (empty) street cross-section. Returns its index.

async openswmm_mcp.tools.infrastructure.add_transect(ctx: fastmcp.Context, session_id: str = 'default', transect_id: str = '') → dict#

Create a new (empty) transect. Returns the assigned zero-based index.

Populate the transect with set_transect_roughness() and add_transect_station() calls.

async openswmm_mcp.tools.infrastructure.add_transect_station(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, station: float = 0.0, elevation: float = 0.0) → dict#: Append a single (station, elevation) point to a transect’s profile.

async openswmm_mcp.tools.infrastructure.clear_stations(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#: Remove all (station, elevation) points from a transect’s profile.

async openswmm_mcp.tools.infrastructure.get_bank_stations(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#: Read back a transect’s left/right bank station positions.

async openswmm_mcp.tools.infrastructure.get_comments(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#: Read back the comment text attached to a transect.

async openswmm_mcp.tools.infrastructure.get_encroachment_stations(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#: Read back a transect’s left/right encroachment station positions.

async openswmm_mcp.tools.infrastructure.get_lid_drain(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0) → dict#

Read LID underdrain parameters. Inverse of set_lid_drain().

Returns coeff, expon, offset.

async openswmm_mcp.tools.infrastructure.get_lid_drainmat(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0) → dict#

Read LID drainage-mat parameters. Inverse of set_lid_drainmat().

Returns thick, void_frac, roughness.

async openswmm_mcp.tools.infrastructure.get_lid_pavement(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0) → dict#

Read LID porous-pavement parameters. Inverse of set_lid_pavement().

Returns thick, void_ratio, frac_imperv, ksat, clog_factor, regen_days.

async openswmm_mcp.tools.infrastructure.get_lid_soil(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0) → dict#

Read LID soil-layer parameters. Inverse of set_lid_soil().

Returns thick, porosity, fc, wp, ksat, kslope.

async openswmm_mcp.tools.infrastructure.get_lid_storage(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0) → dict#

Read LID storage-layer parameters. Inverse of set_lid_storage().

Returns thick, void_frac, ksat.

async openswmm_mcp.tools.infrastructure.get_lid_surface(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0) → dict#

Read LID surface-layer parameters. Inverse of set_lid_surface().

Returns storage, roughness, slope — the same keys set_lid_surface() accepts.

async openswmm_mcp.tools.infrastructure.get_modifiers(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#: Read back a transect’s roughness / station / elevation modifier factors.

async openswmm_mcp.tools.infrastructure.get_station(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, station_index: int = 0) → dict#: Read back a single (station, elevation) point from a transect’s profile.

async openswmm_mcp.tools.infrastructure.get_street_params(ctx: fastmcp.Context, session_id: str = 'default', street_index: int = 0) → dict#

Read back a street cross-section’s geometric parameters.

Inverse of set_street_params(). Returns a params dict with keys t_crown, h_curb, sx, n_road, gutter_depres, gutter_width, sides, back_width, back_slope, back_n.

async openswmm_mcp.tools.infrastructure.get_transect_roughness(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#

Read back the three Manning’s roughness values of a transect.

Inverse of set_transect_roughness(). Returns n_left / n_right (overbank) and n_channel (main channel).

async openswmm_mcp.tools.infrastructure.inlet_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of inlets defined in the model.

async openswmm_mcp.tools.infrastructure.lid_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of LID controls defined in the model.

async openswmm_mcp.tools.infrastructure.lid_usage_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of [LID_USAGE] placement rows across all subcatchments.

async openswmm_mcp.tools.infrastructure.lid_usage_get(ctx: fastmcp.Context, session_id: str = 'default', usage_index: int = 0) → dict#

Read one [LID_USAGE] placement row by global index.

Returns the owning subcatchment/LID indices and the placement parameters (number, area, width, init_sat, from_imperv, to_perv, from_perv).

async openswmm_mcp.tools.infrastructure.lid_usage_remove(ctx: fastmcp.Context, session_id: str = 'default', usage_index: int = 0) → dict#: Remove one [LID_USAGE] placement row by global index.

async openswmm_mcp.tools.infrastructure.remove_transect(ctx: fastmcp.Context, session_id: str = 'default', transect: str | int = '') → dict#: Remove a transect by string ID or integer index.

async openswmm_mcp.tools.infrastructure.set_bank_stations(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, left: float = 0.0, right: float = 0.0) → dict#

Set a transect’s left/right bank station positions.

The bank stations delimit the main channel from the overbank zones (which use the left/right roughness from set_transect_roughness()).

async openswmm_mcp.tools.infrastructure.set_comments(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, text: str = '') → dict#: Set the comment text attached to a transect.

async openswmm_mcp.tools.infrastructure.set_encroachment_stations(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, left: float = 0.0, right: float = 0.0) → dict#: Set a transect’s left/right encroachment station positions.

async openswmm_mcp.tools.infrastructure.set_inlet_params(ctx: fastmcp.Context, session_id: str = 'default', inlet_index: int = 0, length: float = 0.0, width: float = 0.0, grate_type: str = '', open_area: float = 0.0, splash_veloc: float = 0.0) → dict#

Set the operational parameters for an inlet.

grate_type identifies a grate-style family (e.g. "P_BAR-50"); consult the engine for the available identifiers. open_area is the open-area fraction; splash_veloc is the splash-over velocity.

async openswmm_mcp.tools.infrastructure.set_lid_drain(ctx: fastmcp.Context, session_id: str = 'default', lid_index: int = 0, coeff: float = 0.0, expon: float = 0.0, offset: float = 0.0) → dict#

Set LID underdrain parameters: discharge coefficient, exponent, offset.

Drain flow follows Q = coeff * h^expon for head above offset.

async openswmm_mcp.tools.infrastructure.set_lid_drainmat(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0, thick: float = 0.0, void_frac: float = 0.0, roughness: float = 0.0) → dict#

Set LID drainage-mat layer parameters (green_roof LIDs).

lid_index accepts a string LID ID or an integer index. The mat is described by its thickness, void fraction, and Manning’s roughness.

async openswmm_mcp.tools.infrastructure.set_lid_pavement(ctx: fastmcp.Context, session_id: str = 'default', lid_index: str | int = 0, thick: float = 0.0, void_ratio: float = 0.0, frac_imperv: float = 0.0, ksat: float = 0.0, clog_factor: float = 0.0, regen_days: float = 0.0) → dict#

Set LID porous-pavement layer parameters (perm_pavement LIDs).

Parameters:

lid_index – Target LID control (string ID or integer index).
thick – Pavement layer thickness.
void_ratio – Void volume / solids volume (a ratio, not a fraction).
frac_imperv – Impervious surface fraction of the pavement in [0.0, 1.0].
ksat – Saturated hydraulic conductivity of the pavement.
clog_factor – Clogging factor and the pavement regeneration interval in days.
regen_days – Clogging factor and the pavement regeneration interval in days.

async openswmm_mcp.tools.infrastructure.set_lid_soil(ctx: fastmcp.Context, session_id: str = 'default', lid_index: int = 0, thick: float = 0.0, porosity: float = 0.0, fc: float = 0.0, wp: float = 0.0, ksat: float = 0.0, kslope: float = 0.0) → dict#

Set LID soil-layer parameters.

Parameters:

thick – Soil thickness.
porosity – Porosity, field capacity, and wilting point (all volume fractions).
fc – Porosity, field capacity, and wilting point (all volume fractions).
wp – Porosity, field capacity, and wilting point (all volume fractions).
ksat – Saturated hydraulic conductivity.
kslope – Conductivity slope (Green-Ampt parameter).

async openswmm_mcp.tools.infrastructure.set_lid_storage(ctx: fastmcp.Context, session_id: str = 'default', lid_index: int = 0, thick: float = 0.0, void_frac: float = 0.0, ksat: float = 0.0) → dict#: Set LID storage-layer parameters: thickness, void fraction, k_sat.

async openswmm_mcp.tools.infrastructure.set_lid_surface(ctx: fastmcp.Context, session_id: str = 'default', lid_index: int = 0, storage: float = 0.0, roughness: float = 0.0, slope: float = 0.0) → dict#: Set LID surface-layer parameters: storage depth, roughness, slope.

async openswmm_mcp.tools.infrastructure.set_modifiers(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, n_factor: float = 0.0, x_factor: float = 0.0, y_factor: float = 0.0) → dict#

Set a transect’s modifier factors.

n_factor scales roughness, x_factor scales station distances, and y_factor scales elevations.

async openswmm_mcp.tools.infrastructure.set_street_params(ctx: fastmcp.Context, session_id: str = 'default', street_index: int = 0, t_crown: float = 0.0, h_curb: float = 0.0, sx: float = 0.0, n_road: float = 0.0, gutter_depres: float = 0.0, gutter_width: float = 0.0, sides: int = 1, back_width: float = 0.0, back_slope: float = 0.0, back_n: float = 0.0) → dict#

Set the full geometry of a street cross-section.

Parameters:

t_crown – Crown (road centerline) elevation width.
h_curb – Curb height.
sx – Road cross slope (rise/run).
n_road – Manning’s M{n} for the road surface.
gutter_depres – Gutter depression depth and width.
gutter_width – Gutter depression depth and width.
sides – 1 = single-sided (one curb), 2 = double-sided.
back_width – Backing (behind-curb) geometry — width, slope, and Manning’s M{n}.
back_slope – Backing (behind-curb) geometry — width, slope, and Manning’s M{n}.
back_n – Backing (behind-curb) geometry — width, slope, and Manning’s M{n}.

async openswmm_mcp.tools.infrastructure.set_transect_roughness(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0, n_left: float = 0.0, n_right: float = 0.0, n_channel: float = 0.0) → dict#

Set Manning’s roughness values for the three transect zones.

n_left and n_right are the overbank roughness; n_channel is the main channel.

async openswmm_mcp.tools.infrastructure.station_count(ctx: fastmcp.Context, session_id: str = 'default', transect_index: int = 0) → dict#: Return the number of (station, elevation) points in a transect’s profile.

async openswmm_mcp.tools.infrastructure.street_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of street cross-sections in the model.

async openswmm_mcp.tools.infrastructure.transect_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of transects defined in the model.

openswmm_mcp.tools.hotstart#

Hot-start save/load and session-cloning tools.

async openswmm_mcp.tools.hotstart.clone_session(ctx: fastmcp.Context, source_id: str = '', target_id: str = '') → dict#

Clone an existing session by saving and re-applying its hot-start state.

A new session is created using the same .inp file as the source. The source’s current hydraulic state is written to a temporary hot-start file and then applied to the freshly opened target session.

Parameters:

source_id – Identifier of the session to clone.
target_id – Identifier for the new cloned session.

async openswmm_mcp.tools.hotstart.get_file_sim_time(ctx: fastmcp.Context, session_id: str = 'default', path: str = '') → dict#

Return the simulation moment stored inside a hot-start file.

This is the timestamp the state was captured at, read from the file’s header — not the live clock. lifecycle_get_simulation_time reports the running session’s current time; this tool answers “what point in the run does this checkpoint represent?” without applying it to anything.

No session state is touched; session_id is accepted only so the tool is uniform with the rest of the namespace.

Parameters:: path – Path to an existing hot-start file.

async openswmm_mcp.tools.hotstart.load_hotstart(ctx: fastmcp.Context, session_id: str = 'default', path: str = '') → HotStartResult#

Load a previously saved hot-start file into a session.

The hot-start file is opened and its state is applied to the session’s solver, allowing a simulation to resume from a saved checkpoint.

Parameters:

session_id – Identifier of the target session. Defaults to "default".
path – File-system path of the hot-start file to load.

async openswmm_mcp.tools.hotstart.save_hotstart(ctx: fastmcp.Context, session_id: str = 'default', path: str = '') → HotStartResult#

Save the current simulation state to a hot-start file.

The session must be in the "running" or "ended" state so that there is meaningful hydraulic state to persist.

Parameters:

session_id – Identifier of the session to save. Defaults to "default".
path – File-system path for the hot-start file. When empty, a path is generated inside the session’s working directory.

async openswmm_mcp.tools.hotstart.saves_add(ctx: fastmcp.Context, session_id: str = 'default', path: str = '', datetime_oadate: float = 0.0) → dict#

Append a new SAVE HOTSTART entry.

datetime_oadate is decimal days (OADate). Use 0.0 to schedule a save at end of simulation.

async openswmm_mcp.tools.hotstart.saves_clear(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Remove every scheduled save.

async openswmm_mcp.tools.hotstart.saves_count(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Return the number of scheduled SAVE HOTSTART entries in [FILES].

async openswmm_mcp.tools.hotstart.saves_get(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Return the path + datetime of the I{index}-th scheduled save.

async openswmm_mcp.tools.hotstart.saves_remove(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0) → dict#: Remove the I{index}-th scheduled save. Trailing entries shift down.

async openswmm_mcp.tools.hotstart.saves_set(ctx: fastmcp.Context, session_id: str = 'default', index: int = 0, path: str | None = None, datetime_oadate: float | None = None) → dict#

Update the path and/or datetime of the I{index}-th scheduled save.

Fields not supplied (None) are left unchanged. v1 SaveSchedule requires a full entry replacement, so we read-modify-write.

async openswmm_mcp.tools.hotstart.seed_hotstart_state(ctx: fastmcp.Context, session_id: str = 'default', path: str = '', node_depths: dict[str, float] | None = None, node_heads: dict[str, float] | None = None, link_depths: dict[str, float] | None = None, link_flows: dict[str, float] | None = None, subcatchment_runoffs: dict[str, float] | None = None) → HotStartResult#

Seed specific element states from a hot-start file into a session.

Opens the hot-start file at path, overrides individual element states with the supplied id→value maps, and applies the result to the session’s solver. This surfaces the engine’s hot-start state setters (set_node_depth / set_node_head / set_link_depth / set_link_flow / set_subcatchment_runoff) so callers can build deterministic initial conditions — e.g. reproducible RL episode resets.

All values are in the model’s project units (see get_unit_system): depths/heads in project length units, flows in project flow units, runoff in project flow units.

Parameters:

session_id – Target session. Defaults to "default".
path – Hot-start file to open as the seed baseline (required, must exist).
node_depths – Maps of node id → depth / head override.
node_heads – Maps of node id → depth / head override.
link_depths – Maps of link id → depth / flow override.
link_flows – Maps of link id → depth / flow override.
subcatchment_runoffs – Map of subcatchment id → runoff override.

openswmm_mcp.tools.spatial_quality#

Spatial coordinate and water-quality tools.

async openswmm_mcp.tools.spatial_quality.add_lid(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', lid_idx: int = 0, number: int = 1, area: float = 0.0, width: float = 0.0, init_sat: float = 0.0, from_imperv: float = 1.0) → dict#

Add a Low Impact Development (LID) control to a subcatchment.

Parameters:

session_id – Identifier of the session. Defaults to "default".
subcatch_id – The subcatchment that will receive the LID.
lid_idx – Index of the LID control defined in the model (zero-based).
number – Number of identical LID units to place.
area – Surface area of each LID unit (in model area units).
width – Top width of the overland-flow surface of each LID unit.
init_sat – Initial saturation fraction in [0.0, 1.0].
from_imperv – Fraction of impervious area routed to the LID, in [0.0, 1.0].

async openswmm_mcp.tools.spatial_quality.get_all_coordinates(ctx: fastmcp.Context, session_id: str = 'default', element_type: str = 'node') → dict#

Return coordinates for all elements of a given type in one call.

This is much faster than calling get_coordinates() in a loop. For nodes, a single C-level bulk read (memcpy) is used. For links, subcatchments, and gages, all per-element reads are batched inside a single thread call to avoid per-element async overhead.

Returns a list of {"id": "...", "x": ..., "y": ...} records, one per element, in index order.

Parameters:

session_id – Identifier of the session. Defaults to "default".
element_type – One of "node", "link", "subcatchment", or "gage".

async openswmm_mcp.tools.spatial_quality.get_all_polygons(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the boundary polygon for all subcatchments in one call.

Each record contains the subcatchment ID, its centroid, and its full polygon vertex list. All reads are batched inside a single thread to avoid per-subcatchment async overhead.

Returns a list of records:

{"id": "S1", "centroid": [x, y], "vertex_count": 6,
 "polygon": [[x0,y0], ..., [x5,y5]]}

Subcatchments with no polygon geometry return "polygon": [].

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.spatial_quality.get_all_vertices(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the polyline vertices for all links in one call.

Each link entry contains its full ordered polyline (upstream endpoint, any interior shape-points, downstream endpoint). All reads are batched inside a single thread to avoid per-link async overhead.

Returns a list of records:

{"id": "C1", "vertex_count": 3, "vertices": [[x0,y0], [x1,y1], [x2,y2]]}

Links with no stored geometry return "vertices": [].

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.spatial_quality.get_coordinates(ctx: fastmcp.Context, session_id: str = 'default', element_type: str = 'node', element_id: str = '') → SpatialResult#

Retrieve the spatial coordinates for a model element.

Supports nodes (x, y), links (x, y centroid), and subcatchments (x, y centroid) depending on the data stored in the model.

Parameters:

session_id – Identifier of the session. Defaults to "default".
element_type – One of "node", "link", or "subcatchment".
element_id – The identifier of the element whose coordinates are requested.

async openswmm_mcp.tools.spatial_quality.get_crs(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the coordinate reference system (CRS) string for the model.

The CRS is stored as a string such as an EPSG code (e.g. EPSG:4326), a PROJ string, or a WKT string. An empty string means no CRS has been assigned.

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.spatial_quality.get_model_geometry(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return complete geometry for the entire model in a single call.

This is the primary endpoint for rendering a SWMM model. It returns all spatial data needed to draw nodes, links (pipes/channels), subcatchments (watersheds), and rain gages — plus the model CRS and bounding box.

All data is fetched with minimal thread hops:

Nodes use a single C-level bulk memcpy for coordinates.
Links, subcatchments, and gages batch all per-element reads inside one thread call each.

Response structure:

{
  "session_id": "default",
  "crs": "EPSG:4326",
  "bounds": {"min_x": ..., "min_y": ..., "max_x": ..., "max_y": ...},
  "nodes": [
    {"id": "J1", "type": 0, "type_name": "JUNCTION", "x": 0.0, "y": 0.0}
  ],
  "links": [
    {"id": "C1", "type": 0, "type_name": "CONDUIT",
     "from_node": "J1", "to_node": "J2",
     "vertices": [[x0,y0], [x1,y1], ...]}
  ],
  "subcatchments": [
    {"id": "S1", "centroid": [x, y],
     "polygon": [[x0,y0], ...], "outlet_node_idx": 0}
  ],
  "gages": [
    {"id": "RG1", "x": 0.0, "y": 0.0}
  ]
}

Parameters:: session_id – Identifier of the session. Defaults to "default".

async openswmm_mcp.tools.spatial_quality.get_polygon(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '') → dict#

Return the polygon vertices of a subcatchment (watershed boundary).

Vertices define the closed boundary polygon of the subcatchment in model coordinates. An empty list means no polygon geometry has been assigned.

Parameters:

session_id – Identifier of the session. Defaults to "default".
subcatch_id – String identifier of the subcatchment whose polygon is requested.

async openswmm_mcp.tools.spatial_quality.get_quality(ctx: fastmcp.Context, session_id: str = 'default', element_type: str = 'node', element_id: str = '', pollutant: str | None = None) → dict#

Retrieve water-quality concentrations for a model element.

When pollutant is None all tracked pollutants are returned; otherwise only the named pollutant’s concentration is included.

Parameters:

session_id – Identifier of the session. Defaults to "default".
element_type – One of "node", "link", or "subcatchment".
element_id – The identifier of the element to query.
pollutant – Optional pollutant name. When omitted, all pollutants are returned.

async openswmm_mcp.tools.spatial_quality.get_vertices(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '') → dict#

Return the ordered polyline vertices for a link (pipe or channel).

Vertices are returned in upstream-to-downstream order. For conduits with no interior vertices only the two endpoint coordinates (from the connected nodes) are returned.

Parameters:

session_id – Identifier of the session. Defaults to "default".
link_id – String identifier of the link whose vertices are requested.

async openswmm_mcp.tools.spatial_quality.set_coordinates(ctx: fastmcp.Context, session_id: str = 'default', element_type: str = 'node', element_id: str = '', x: float = 0.0, y: float = 0.0) → dict#

Set the spatial coordinates for a model element.

Parameters:

session_id – Identifier of the session. Defaults to "default".
element_type – One of "node", "link", or "subcatchment".
element_id – The identifier of the element to update.
x – The new X coordinate.
y – The new Y coordinate.

async openswmm_mcp.tools.spatial_quality.set_crs(ctx: fastmcp.Context, session_id: str = 'default', crs: str = '') → dict#

Set the coordinate reference system (CRS) string for the model.

Parameters:

session_id – Identifier of the session. Defaults to "default".
crs – CRS identifier string, e.g. "EPSG:4326", a PROJ string, or WKT.

async openswmm_mcp.tools.spatial_quality.set_gage_coord(ctx: fastmcp.Context, session_id: str = 'default', gage_id: str = '', x: float = 0.0, y: float = 0.0) → dict#

Set the (x, y) symbol coordinates of a rain gage.

Complements get_all_coordinates() (which reads gage coordinates via element_type="gage"). The element-keyed set_coordinates() tool handles nodes / links / subcatchments only; this is the dedicated gage-coordinate setter.

Parameters:

session_id – Identifier of the session. Defaults to "default".
gage_id – The identifier of the rain gage to update.
x – The new X coordinate.
y – The new Y coordinate.

async openswmm_mcp.tools.spatial_quality.set_node_coords_bulk(ctx: fastmcp.Context, session_id: str = 'default', coordinates: list[list[float]] = []) → dict#

Set all node coordinates in one bulk call.

The inverse of reading node coordinates via get_all_coordinates() (element_type="node"). coordinates must be a list of [x, y] pairs in node-index order, one per node in the model; a single C-level bulk write (memcpy) replaces every node’s coordinates at once.

Parameters:

session_id – Identifier of the session. Defaults to "default".
coordinates – List of [x, y] pairs, one per node, in index order.

async openswmm_mcp.tools.spatial_quality.set_polygon(ctx: fastmcp.Context, session_id: str = 'default', subcatch_id: str = '', polygon: list[list[float]] = []) → dict#

Set the polygon boundary of a subcatchment (watershed).

Replaces any existing polygon geometry. Each entry in polygon must be a two-element list [x, y]. Pass an empty list to clear the polygon.

Parameters:

session_id – Identifier of the session. Defaults to "default".
subcatch_id – String identifier of the subcatchment to update.
polygon – Ordered list of [x, y] coordinate pairs defining the closed boundary.

async openswmm_mcp.tools.spatial_quality.set_treatment(ctx: fastmcp.Context, session_id: str = 'default', node_id: str = '', pollutant: str = '', expression: str = '') → dict#

Assign a treatment expression to a node for a given pollutant.

Treatment expressions use SWMM’s built-in syntax (e.g. "R = 0.5 * C" to remove 50 % of concentration C).

Parameters:

session_id – Identifier of the session. Defaults to "default".
node_id – The node to which the treatment applies.
pollutant – Name of the pollutant being treated.
expression – SWMM treatment expression string.

async openswmm_mcp.tools.spatial_quality.set_vertices(ctx: fastmcp.Context, session_id: str = 'default', link_id: str = '', vertices: list[list[float]] = []) → dict#

Set the ordered polyline vertices for a link.

Replaces any existing interior vertices. Each entry in vertices must be a two-element list [x, y]. Pass an empty list to clear all interior vertices.

Parameters:

session_id – Identifier of the session. Defaults to "default".
link_id – String identifier of the link to update.
vertices – Ordered list of [x, y] coordinate pairs, upstream to downstream.

openswmm_mcp.tools.geopackage#

GeoPackage Tools#

Tools for querying SWMM GeoPackage databases — simulation results, observed data import/export, and multi-run scenario comparison.

author:: Caleb Buahin
license:: Apache-2.0

async openswmm_mcp.tools.geopackage.close_geopackage(ctx: fastmcp.Context, session_id: str = 'gpkg_default') → dict#

Close a GeoPackage connection.

Parameters:: session_id – GeoPackage session identifier.

async openswmm_mcp.tools.geopackage.compare_sim_vs_observed(ctx: fastmcp.Context, session_id: str = 'gpkg_default', simulation_id: str = '', element_type: str = 'NODE', element_id: str = '', variable: str = 'depth', observed_series_id: int = 0) → dict#

Compare simulated results against observed data.

Parameters:

session_id – GeoPackage session identifier.
simulation_id – Simulation run to compare.
element_type – Element type.
element_id – Element ID.
variable – Variable to compare.
observed_series_id – Observed series ID.

Returns:

Comparison statistics (RMSE, NSE, bias).

async openswmm_mcp.tools.geopackage.get_result_summary(ctx: fastmcp.Context, session_id: str = 'gpkg_default', simulation_id: str = '', element_type: str = 'NODE', element_id: str = '', variable: str = 'max_depth') → dict#

Read a summary statistic from the GeoPackage.

Parameters:

session_id – GeoPackage session identifier.
simulation_id – Simulation run ID.
element_type – “NODE”, “LINK”, or “SUBCATCH”.
element_id – Element identifier.
variable – Summary variable (e.g., “max_depth”, “max_flow”).

Returns:

Dict with the statistic value.

async openswmm_mcp.tools.geopackage.get_result_timeseries(ctx: fastmcp.Context, session_id: str = 'gpkg_default', simulation_id: str = '', element_type: str = 'NODE', element_id: str = '', variable: str = 'depth') → dict#

Read a simulation result timeseries from the GeoPackage.

Parameters:

session_id – GeoPackage session identifier.
simulation_id – Simulation run ID (use list_simulations to find).
element_type – “NODE”, “LINK”, “SUBCATCH”, or “SYSTEM”.
element_id – Element identifier (e.g., “J1”).
variable – Variable name (e.g., “depth”, “flow”, “runoff”).

Returns:

Dict with times and values arrays.

async openswmm_mcp.tools.geopackage.import_observed_data(ctx: fastmcp.Context, session_id: str = 'gpkg_default', name: str = '', variable: str = 'flow', element_type: str = '', element_id: str = '', timestamps: list[str] | None = None, values: list[float] | None = None, source: str = '', units: str = '') → dict#

Import observed/sensor data into the GeoPackage for calibration.

Parameters:

session_id – GeoPackage session identifier.
name – Unique series name (e.g., “USGS_01585200_flow”).
variable – Variable measured (e.g., “flow”, “depth”).
element_type – Model element type to link to (or “” for unlinked).
element_id – Model element ID to link to (or “”).
timestamps – List of ISO 8601 timestamp strings.
values – List of measured values.
source – Data source description.
units – Measurement units.

Returns:

Dict with series_id and count.

async openswmm_mcp.tools.geopackage.is_registered(ctx: fastmcp.Context) → dict#

Check whether the GeoPackage plugin is registered.

Returns:: Dict with the boolean registered flag.

async openswmm_mcp.tools.geopackage.last_error(ctx: fastmcp.Context, session_id: str = 'gpkg_default') → dict#

Return the most recent error message from the GeoPackage library.

Parameters:: session_id – GeoPackage session identifier.
Returns:: Dict with the last_error string (”” if none).

async openswmm_mcp.tools.geopackage.list_simulations(ctx: fastmcp.Context, session_id: str = 'gpkg_default') → list[dict]#

List all simulation runs in the GeoPackage.

Parameters:: session_id – GeoPackage session identifier.
Returns:: List of simulation metadata dicts.

async openswmm_mcp.tools.geopackage.open_geopackage(ctx: fastmcp.Context, path: str, session_id: str = 'gpkg_default') → dict#

Open a GeoPackage file for querying results or observed data.

Parameters:

path – Path to the .gpkg file.
session_id – Session identifier for this GeoPackage connection.

Returns:

Summary of the GeoPackage contents.

async openswmm_mcp.tools.geopackage.query_double(ctx: fastmcp.Context, session_id: str = 'gpkg_default', sql: str = '') → dict#

Run a read-only SQL query and return the first double result.

Parameters:

session_id – GeoPackage session identifier.
sql – SELECT query string.

Returns:

Dict with the float value.

async openswmm_mcp.tools.geopackage.query_int(ctx: fastmcp.Context, session_id: str = 'gpkg_default', sql: str = '') → dict#

Run a read-only SQL query and return the first integer result.

Parameters:

session_id – GeoPackage session identifier.
sql – SELECT query string.

Returns:

Dict with the integer value.

async openswmm_mcp.tools.geopackage.register(ctx: fastmcp.Context, key: str = '', org: str = '', email: str = '', deploy: str = '') → dict#

Parameters:

key – License key, or “” if not required.
org – Organisation name, or “”.
email – Contact e-mail, or “”.
deploy – Deployment identifier, or “”.

Returns:

Dict with the boolean registered result.

async openswmm_mcp.tools.geopackage.topology_edge_count(ctx: fastmcp.Context, session_id: str = 'gpkg_default', simulation_id: str = '') → dict#

Return the number of topology edges for a simulation.

Parameters:

session_id – GeoPackage session identifier.
simulation_id – Simulation run ID.

Returns:

Dict with the integer edge_count.

async openswmm_mcp.tools.geopackage.write_observed_value(ctx: fastmcp.Context, session_id: str = 'gpkg_default', series_id: int = 0, timestamp: str = '', value: float = 0.0, flag: str = '') → dict#

Write a single observed data point to an existing series.

Use import_observed_data to create a series and bulk-load points; this tool appends one point to a series that already exists.

Parameters:

session_id – GeoPackage session identifier.
series_id – Series ID returned by import_observed_data.
timestamp – ISO 8601 timestamp string.
value – Measured value.
flag – Quality flag (e.g. “A”, “P”), or “” for none.

Returns:

Dict confirming the write.

openswmm_mcp.tools.twod#

2D overland-flow surface tools for the OpenSWMM MCP server.

Surfaces the engine’s Surface2D view (openswmm.engine._2d) — mesh queries, per-triangle state, statistics, mass balance, runtime forcing, edge boundary conditions, and edge conveyance — for models that carry [2D_*] sections (or an external mesh file).

All tools require the new openswmm engine backend and an OPENED or later session. Bulk per-triangle results are returned as summary statistics plus an optional offset / limit slice so responses stay LLM-friendly on large meshes.

async openswmm_mcp.tools.twod.add_triangle_coupling(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, node_name: str = '', cd: float = 0.65, area: float = 1.0) → dict#

Couple a 2D mesh triangle to a 1D SWMM node (node->cell exchange).

Appends one [2D_TRIANGLE_NODE_MAP] row; it does not overwrite, so a triangle may carry several rows (one per node). cd is the discharge coefficient (> 0, default 0.65) and area the effective exchange area in m2 (> 0, default 1.0). Use twod_clear_triangle_couplings to re-author the whole set.

async openswmm_mcp.tools.twod.clear_triangle_couplings(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Remove every authored triangle (node->cell) coupling row.

Also resets the legacy per-triangle mirror read by twod_get_coupling_map. Vertex couplings are untouched.

async openswmm_mcp.tools.twod.force_clear(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Clear every 2D forcing override (rainfall and coupling flux).

async openswmm_mcp.tools.twod.force_coupling_flux(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, value: float = 0.0, mode: str = 'replace', persist: bool = False) → dict#

Force the 1D<->2D coupling flux on a triangle (m/s, positive = into 2D).

mode is "replace" or "add"; persist=True holds the forcing until cleared, otherwise it resets after one step.

async openswmm_mcp.tools.twod.force_evap(ctx: fastmcp.Context, session_id: str = 'default', value: float = 0.0, triangle: int = -1, mode: str = 'replace', persist: bool = False) → dict#

Force evaporation on the 2D surface (m/s).

triangle < 0 (the default) applies the rate uniformly to every triangle; otherwise only the given triangle is forced. mode is "replace" or "add"; persist=True holds the forcing until cleared, otherwise it resets after one step.

async openswmm_mcp.tools.twod.force_rainfall(ctx: fastmcp.Context, session_id: str = 'default', value: float = 0.0, triangle: int = -1, mode: str = 'replace', persist: bool = False) → dict#

Force rainfall on the 2D surface (m/s).

async openswmm_mcp.tools.twod.get_coupling_map(ctx: fastmcp.Context, session_id: str = 'default') → dict#

List every 2D mesh entity coupled to a 1D node.

Returns vertex_couplings (vertex index -> node index) and triangle_couplings (triangle index -> node index). These are the exchange points where the 2D surface trades flow with the drainage network.

triangle_coupling_rows is the authoritative [2D_TRIANGLE_NODE_MAP] row list — {row, triangle, node_index, cd, area} — and is what twod_add_triangle_coupling writes. Prefer it over triangle_couplings, which is a lossy per-triangle mirror showing only one node per triangle.

async openswmm_mcp.tools.twod.get_edge_bc(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, edge: int = 0) → dict#

Return the boundary condition on one triangle edge.

Reports the BC type (WALL / NORMAL_FLOW / SPECIFIED_STAGE / SPECIFIED_FLOW / RATING_CURVE), the constant head and slope, the prescribed per-metre flow, and the cumulative flux through the edge.

async openswmm_mcp.tools.twod.get_edge_conveyance(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = -1, edge: int = 0) → dict#

Read per-edge conveyance factors (1.0 = unrestricted, 0.0 = wall).

With triangle >= 0 returns the single factor at (triangle, edge); with triangle < 0 (default) returns a whole-mesh summary plus the list of restricted edges (factor < 1), so berms / barriers are easy to spot.

async openswmm_mcp.tools.twod.get_edge_geometry_bulk(ctx: fastmcp.Context, session_id: str = 'default', offset: int = 0, limit: int = 0) → dict#

Return time-invariant per-edge geometry for the whole mesh.

For every triangle edge (indexed [tri*3 + edge]) reports its length (m) and the outward unit-normal components nx / ny. Returns summary statistics for each array plus the per-edge values in [offset, offset+limit) when limit > 0, each entry as {triangle, edge, length, nx, ny}. Pairs with twod_get_state_bulk (variable="edge_flux"), which shares the same [tri*3 + edge] indexing.

async openswmm_mcp.tools.twod.get_mass_balance(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the global 2D mass-balance terms (m3) and continuity error.

Terms: initial/final storage, rainfall in, 1D->2D coupling in, 2D->1D coupling out, outfall in/out, evaporation out, boundary in/out, and the overall continuity error as a fraction of total inflow.

async openswmm_mcp.tools.twod.get_mesh_geometry(ctx: fastmcp.Context, session_id: str = 'default', offset: int = 0, limit: int = 50) → dict#

Return mesh geometry for a window of triangles.

For each triangle in [offset, offset+limit): its vertex indices, area, centroid, Manning’s n, and the three neighbour triangle indices (-1 = boundary). Also reports vertex elevation summary statistics. Use twod_get_mesh_summary first to learn the mesh size.

async openswmm_mcp.tools.twod.get_mesh_summary(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Report whether the model has an active 2D surface and its mesh sizes.

Returns active, vertex / triangle counts, the number of boundary edges, and how many vertices / triangles are coupled to 1D nodes. Safe to call on any opened model — active is False when the model carries no [2D_*] sections.

async openswmm_mcp.tools.twod.get_solver_params(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return the 2D solver parameters.

Reports the dry-depth threshold (m). The explicit-marcher configuration (THETA, CFL_NUMBER, LTS_TIERS, H_MOVE, FROUDE_MAX, COUPLING_AREA, …) lives in [2D_OPTIONS] and is read with model_get_option_ext. The retired CVODE tolerances no longer exist.

async openswmm_mcp.tools.twod.get_state(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0) → dict#

Return the current 2D state at one triangle.

Reports depth (m), head (m), rainfall (m/s), net source (m/s), and the 1D<->2D coupling flux (m3/s, positive = into the 2D surface).

async openswmm_mcp.tools.twod.get_state_bulk(ctx: fastmcp.Context, session_id: str = 'default', variable: str = 'depth', offset: int = 0, limit: int = 0) → dict#

Summarise a 2D state variable over the whole mesh.

variable is one of "depth", "head", "vertex_head", "vertex_render_depth", "coupling_flux", or "edge_flux". Returns count / min / max / mean, plus the values in [offset, offset+limit) when limit > 0 (edge_flux and the others are indexed per triangle except vertex_head and vertex_render_depth, which are per vertex; edge_flux is [tri*3 + edge]). vertex_render_depth is the render-oriented signed vertex water depth (eta_v - z_v) GUIs should interpolate for 2D water-surface rendering.

async openswmm_mcp.tools.twod.get_stats(ctx: fastmcp.Context, session_id: str = 'default', top_n: int = 10) → dict#

Return cumulative per-triangle statistics with worst-case hot spots.

For max depth (m), max velocity magnitude (m/s), and max absolute continuity residual (m3/s): summary statistics plus the top_n triangles with the largest values (index + value), ranked descending.

async openswmm_mcp.tools.twod.get_totals(ctx: fastmcp.Context, session_id: str = 'default') → dict#

Return whole-surface totals and internal-stepper diagnostics.

Reports max depth over the surface (m), total ponded volume (m3), total 1D<->2D exchange flow (m3/s), the explicit marcher’s sub-step count for the last advance, and its last sub-step size (s).

async openswmm_mcp.tools.twod.get_triangle_initial_conditions(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0) → dict#

Return the initial water depth and velocity of a 2D triangle.

init_depth is the [2D_TRIANGLES] INIT_DEPTH column in mesh length units — feet on a US-FLOW_UNITS project, metres on an SI project (the same convention as the vertex Z column), not the SI metres used by the run-time state tools. init_u / init_v are the [2D_INITIAL_VELOCITY] components and are always in m/s.

async openswmm_mcp.tools.twod.get_triangle_tag(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0) → dict#: Return the descriptive tag of a 2D triangle (empty if untagged).

async openswmm_mcp.tools.twod.get_vertex_coupling_params(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0) → dict#

Return the 1D<->2D exchange parameters of a mesh vertex.

Reports the [2D_VERTEX_NODE_MAP] CD (discharge coefficient, default 0.65) and AREA (effective exchange area in m2, default 1.0) columns. Use twod_get_coupling_map for which node each vertex is coupled to.

async openswmm_mcp.tools.twod.get_vertex_head(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0) → dict#

Return the reconstructed water-surface head (m) at one mesh vertex.

Triangle-based state is the solver’s native representation; this is the per-vertex value reconstructed for rendering. Use twod_get_state_bulk with variable="vertex_head" to read every vertex at once.

async openswmm_mcp.tools.twod.get_vertex_tag(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0) → dict#: Return the descriptive tag of a 2D vertex (empty if untagged).

async openswmm_mcp.tools.twod.reset_edge_conveyance(ctx: fastmcp.Context, session_id: str = 'default') → dict#: Reset every edge’s conveyance factor to 1.0 (unrestricted).

async openswmm_mcp.tools.twod.set_edge_bc(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, edge: int = 0, bc_type: str = '', head: float | None = None, slope: float | None = None, flow: float | None = None, tseries_name: str | None = None, flow_tseries_name: str | None = None, rating_curve_name: str | None = None) → dict#

Configure the boundary condition on one triangle edge.

bc_type (optional) is WALL, NORMAL_FLOW, SPECIFIED_STAGE, SPECIFIED_FLOW, or RATING_CURVE. The remaining parameters apply only when provided: head (constant stage, m), slope (NORMAL_FLOW bed slope), flow (per-metre discharge, m3/s/m), tseries_name (stage timeseries; “” clears), flow_tseries_name (flow timeseries; “” clears), rating_curve_name (stage-to-flow curve; “” clears).

async openswmm_mcp.tools.twod.set_edge_conveyance(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, edge: int = 0, conveyance: float = 1.0) → dict#

Set the conveyance factor on one triangle edge (in [0, 1]).

0.0 makes the edge a wall; 1.0 leaves it unrestricted. Interior edges mirror the value to the neighbouring triangle’s partner slot so mass conservation is preserved. Apply between routing steps.

async openswmm_mcp.tools.twod.set_solver_params(ctx: fastmcp.Context, session_id: str = 'default', dry_depth: float | None = None) → dict#

Set 2D solver parameters; omitted parameters are left unchanged.

dry_depth is the wet/dry threshold (m). The explicit-marcher configuration (THETA, CFL_NUMBER, LTS_TIERS, H_MOVE, FROUDE_MAX, COUPLING_AREA, …) lives in [2D_OPTIONS] and is set with model_set_option_ext. The retired CVODE tolerances no longer exist.

async openswmm_mcp.tools.twod.set_triangle_initial_conditions(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, depth: float | None = None, u: float | None = None, v: float | None = None) → dict#

Set the initial depth and/or velocity of a 2D triangle.

depth (>= 0) is in mesh length units — feet on a US-FLOW_UNITS project, metres on an SI project, matching the vertex Z column — and persists in the INIT_DEPTH column of [2D_TRIANGLES]. u and v are the initial velocity components in m/s and must be given together; they persist as [2D_INITIAL_VELOCITY] rows. Both are applied when the 2D surface initializes (t = 0 only — a hotstart still zeroes face momentum), so set them before the run starts.

async openswmm_mcp.tools.twod.set_triangle_mannings(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, n: float = 0.0) → dict#

Set Manning’s roughness for a 2D mesh triangle (must be > 0).

Persists in the MANNINGS_N column of [2D_TRIANGLES] on save.

async openswmm_mcp.tools.twod.set_triangle_tag(ctx: fastmcp.Context, session_id: str = 'default', triangle: int = 0, tag: str = '') → dict#

Set the descriptive tag of a 2D triangle ([2D_TRIANGLES] TAG).

An empty string clears the tag.

async openswmm_mcp.tools.twod.set_vertex_coupled_node(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0, node_name: str = '') → dict#

Couple a 2D mesh vertex to a 1D SWMM node by name.

Establishes the per-vertex 1D<->2D exchange point. Pass an empty string to clear the coupling.

async openswmm_mcp.tools.twod.set_vertex_coupling_params(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0, cd: float | None = None, area: float | None = None) → dict#

Set the 1D<->2D exchange parameters of a mesh vertex.

cd is the orifice/weir discharge coefficient (must be > 0, engine default 0.65) and area the effective exchange area in m2 (must be > 0, default 1.0). Omitted parameters are left unchanged. Both persist in [2D_VERTEX_NODE_MAP]; pair with twod_set_vertex_coupled_node, which establishes the coupling itself.

async openswmm_mcp.tools.twod.set_vertex_tag(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0, tag: str = '') → dict#

Set the descriptive tag of a 2D vertex ([2D_VERTICES] TAG).

An empty string clears the tag. Distinct from the 1D<->2D coupling node.

async openswmm_mcp.tools.twod.set_vertex_z(ctx: fastmcp.Context, session_id: str = 'default', vertex: int = 0, z: float = 0.0) → dict#

Set the ground elevation of a mesh vertex (m).

Updates derived geometry for every triangle incident to the vertex. Useful for what-if terrain edits (berms, regrading) between steps.

Resource Modules #

openswmm_mcp.resources.model#

MCP resources exposing SWMM model data via swmm:// URIs.

async openswmm_mcp.resources.model.link_detail(session_id: str, link_id: str, ctx: fastmcp.Context) → str#: Return a JSON object with full properties and state for a single link.

async openswmm_mcp.resources.model.list_links(session_id: str, ctx: fastmcp.Context) → str#

Return a JSON array of all link IDs with their types.

Phase 4: same bulk pattern as list_nodes.

async openswmm_mcp.resources.model.list_nodes(session_id: str, ctx: fastmcp.Context) → str#

Return a JSON array of all node IDs with their types.

Phase 4: ids come from session.meta.node_ids (cached on first access via the Phase 3 get_ids_bulk accessor); the type field still goes through the per-element scalar accessor but the entire loop runs in a single worker thread instead of N round-trips.

async openswmm_mcp.resources.model.list_sessions(ctx: fastmcp.Context) → str#

Return a JSON array describing every active session.

Each entry contains the session id, current state, and working directory.

async openswmm_mcp.resources.model.list_subcatchments(session_id: str, ctx: fastmcp.Context) → str#

Return a JSON array of all subcatchment IDs.

Phase 4: ids come from session.meta.subcatch_ids — a single C call via the Phase 3 get_ids_bulk accessor (cached for the session’s lifetime). The previous implementation issued N asyncio.to_thread calls.

async openswmm_mcp.resources.model.mass_balance(session_id: str, ctx: fastmcp.Context) → str#: Return a JSON object with continuity errors and volumetric totals.

async openswmm_mcp.resources.model.node_detail(session_id: str, node_id: str, ctx: fastmcp.Context) → str#: Return a JSON object with full properties and state for a single node.

async openswmm_mcp.resources.model.session_summary(session_id: str, ctx: fastmcp.Context) → str#

Return a JSON object summarising the model loaded in session_id.

Includes element counts, flow units, routing model, and time range.

async openswmm_mcp.resources.model.simulation_options(session_id: str, ctx: fastmcp.Context) → str#: Return a JSON object with the model’s simulation options.

Prompt Modules #

openswmm_mcp.prompts.workflows#

Guided workflow prompts for common SWMM modelling tasks.

openswmm_mcp.prompts.workflows.analyze_model(inp_path: str) → str#

Return a comprehensive model-review prompt.

The generated instructions tell the LLM to open the model, inspect its structure, and produce a written assessment covering network topology, hydrology settings, hydraulic parameters, and potential issues.

Parameters:: inp_path – File-system path to the SWMM .inp file to analyse.

openswmm_mcp.prompts.workflows.build_simple_model(description: str) → str#

Return a guided model-construction prompt.

Parameters:: description – Plain-language description of the drainage network to build (e.g. "three subcatchments draining to a single outfall").

openswmm_mcp.prompts.workflows.compare_scenarios(session_a: str, session_b: str) → str#

Return a cross-scenario comparison prompt.

Parameters:

session_a – Identifier of the first (baseline) session.
session_b – Identifier of the second (alternative) session.

openswmm_mcp.prompts.workflows.design_review(session_id: str = 'default', standard: str | None = None) → str#

Return a design-standards-check prompt.

Parameters:

session_id – Session to review.
standard – Optional name of the design standard or regulatory framework to check against (e.g. "10-year" or "local municipal code").

openswmm_mcp.prompts.workflows.diagnose_flooding(session_id: str = 'default', node_ids: str | None = None) → str#

Return a flooding-investigation prompt.

If node_ids is provided (comma-separated), the investigation is scoped to those specific nodes; otherwise the LLM should discover flooded nodes from the simulation results.

Parameters:

session_id – Session to investigate.
node_ids – Optional comma-separated list of node IDs to focus on.

openswmm_mcp.prompts.workflows.explain_results(session_id: str = 'default') → str#

Return a plain-language result-interpretation prompt.

Parameters:: session_id – The session whose results should be explained.

openswmm_mcp.prompts.workflows.what_if(session_id: str = 'default', description: str = '') → str#

Return a what-if scenario setup prompt.

The generated instructions guide the LLM through cloning a session, applying modifications, re-running, and comparing results.

Parameters:

session_id – The baseline session to branch from.
description – Plain-language description of the scenario the user wants to test.

API Reference#

Core Modules#

openswmm_mcp#

openswmm_mcp.server#

openswmm_mcp.config#

openswmm_mcp.session#

openswmm_mcp.dependencies#

openswmm_mcp.models#

openswmm_mcp.errors#

openswmm_mcp.auth#

Authentication strategy#

Backend Modules#

openswmm_mcp.backends.base#

openswmm_mcp.backends.openswmm#

openswmm_mcp.backends.legacy#

Utility Modules#

openswmm_mcp._util.formatting#

openswmm_mcp._util.validation#

Tool Modules#

openswmm_mcp.tools.lifecycle#

openswmm_mcp.tools.query#

openswmm_mcp.tools.forcing#

openswmm_mcp.tools.controls#

openswmm_mcp.tools.analysis#

openswmm_mcp.tools.building#

openswmm_mcp.tools.editing#

openswmm_mcp.tools.model#

openswmm_mcp.tools.nodes#

openswmm_mcp.tools.links#

openswmm_mcp.tools.subcatchments#

openswmm_mcp.tools.inflows#

openswmm_mcp.tools.pollutants#

openswmm_mcp.tools.quality#

openswmm_mcp.tools.tables#

openswmm_mcp.tools.infrastructure#

openswmm_mcp.tools.hotstart#

openswmm_mcp.tools.spatial_quality#

openswmm_mcp.tools.geopackage#

GeoPackage Tools#

openswmm_mcp.tools.twod#

Resource Modules#

openswmm_mcp.resources.model#

Prompt Modules#

openswmm_mcp.prompts.workflows#

Core Modules #

Backend Modules #

Utility Modules #

Tool Modules #

Resource Modules #

Prompt Modules #