REPL API Reference

The REPL module is the core of LeanFlow. It manages the lifecycle of local Lean 4 processes, handles toolchain management (via elan), and provides methods to execute code and query the proof state.

It offers two interfaces:

SyncREPL: A blocking wrapper, ideal for scripts and notebooks.
REPL: The native asynchronous implementation, ideal for high-concurrency workloads.

SynchronousAsynchronous

SyncREPL is a wrapper around REPL abstracting the asyncio event loop.

`leanflow.SyncREPL`

Synchronous wrapper around REPL for convenience.

Provides a blocking API for interacting with the Lean REPL without needing to use async/await.

Example

from leanflow import SyncREPL

with SyncREPL(lean_version="v4.24.0") as repl:
    result = repl.run("theorem add_zero_nat (n : Nat) : n + 0 = n := by sorry")
    print(result)

`init(lean_version=None, repl_path=None, project_path=None, **kwargs)`

Initialize the SyncREPL with the given configuration.

Parameters:

Name	Type	Description	Default
`lean_version`	`Optional[str]`	The Lean version to use (e.g. "4.24.0").	`None`
`repl_path`	`Optional[Path]`	Path to the Lean REPL executable.	`None`
`project_path`	`Optional[Path]`	Path to a local Lean project.	`None`
`**kwargs`	`Any`	Other arguments passed to REPL (timeout, max_memory_mb).	`{}`

`start()`

Start the REPL process.

This is called automatically when using the context manager.

`run(command, env=None)`

Run a Lean command synchronously.

Parameters:

Name	Type	Description	Default
`command`	`str`	A single command string.	required
`env`	`Optional[Union[Environment, int]]`	Optional environment to run in.	`None`

Returns:

Type	Description
`Union[Environment, LeanError]`	Environment or LeanError.

`run_list(commands, env=None)`

Run Lean commands synchronously.

Parameters:

Name	Type	Description	Default
`commands`	`list[str]`	A list of command strings.	required
`env`	`Optional[Union[Environment, int]]`	Optional environment to run in.	`None`

Returns:

Type	Description
`list[Union[Environment, LeanError]]`	List of results (Environment or LeanError).

`run_file(path, return_all_states=False)`

Run a Lean file synchronously.

Parameters:

Name	Type	Description	Default
`path`	`Union[str, Path]`	Path to the Lean file.	required
`return_all_states`	`bool`	Whether to return all intermediate states.	`False`

Returns:

Type	Description
`Union[Environment, LeanError]`	Environment or LeanError.

`run_tactic(tactic, state)`

Run a tactic in a proof state synchronously.

Parameters:

Name	Type	Description	Default
`tactic`	`str`	The tactic to run.	required
`state`	`ProofState`	The current proof state.	required

Returns:

Type	Description
`Union[ProofState, LeanError]`	New ProofState or LeanError.

`close()`

Close the REPL process and clean up resources.

This is called automatically when using the context manager.

`leanflow.REPL`

REPL is an asynchronous, robust wrapper for the Lean REPL (https://github.com/leanprover-community/repl). It automatically manages its memory and restarts itself.

Run as an async context manager (async with) or to manually call close_async when done to ensure proper cleanup.

Example

import asyncio
from leanflow import REPL

async def main():
    async with REPL(lean_version="4.24.0") as repl:
        result = await repl.run("theorem add_zero_nat (n : Nat) : n + 0 = n := by sorry")
        print(result)

asyncio.run(main())

`init(lean_version=None, repl_path=None, project_path=None, timeout=300, header=None, dacite_config=None, fail_on_header_error=True, log_dir=None, log_level=None, debug=False, manage_memory=True, max_memory_mb=8192, max_restart_attempts=3, **kwargs)`

Initializes the REPL.

Parameters:

Name	Type	Description	Default
`lean_version`	`Optional[str]`	The Lean version to use (e.g. "4.21.0").	`None`
`repl_path`	`Optional[Path]`	Path to the Lean REPL executable.	`None`
`project_path`	`Optional[Path]`	Path to the Lean project directory.	`None`
`timeout`	`int`	Timeout for the REPL process.	`300`
`header`	`Optional[str]`	Header to be sent to the REPL.	`None`
`dacite_config`	`Optional[Config]`	Configuration for dacite.	`None`
`fail_on_header_error`	`bool`	If True, raises LeanHeaderError when header execution fails. If False, log the error and continue with base_env_id=None.	`True`
`log_dir`	`Optional[Union[str, Path]]`	Explicit directory to save log files.	`None`
`log_level`	`Optional[str]`	The logging level (e.g., "INFO", "DEBUG").	`None`
`debug`	`bool`	If True, enables file logging even if log_dir is not provided.	`False`
`manage_memory`	`bool`	If True, manages memory and restarts if needed.	`True`
`max_memory_mb`	`int`	Maximum memory in MB before a restart is triggered.	`8192`
`max_restart_attempts`	`int`	Maximum consecutive restarts before raising MemoryError.	`3`
`**kwargs`	`Any`	Additional keyword arguments passed to the EnvironmentManager.	`{}`

`load_from_yaml(yaml_path)` `classmethod`

Loads a REPL instance from a YAML file.

Parameters:

Name	Type	Description	Default
`yaml_path`	`Path`	Path to the YAML file.	required

Returns:

Type	Description
`REPL`	The loaded REPL instance.

`run(command, env=None)` `async`

Runs a Lean command in the specified environment.

Parameters:

Name	Type	Description	Default
`command`	`str`	The Lean command to execute.	required
`env`	`Optional[Union[Environment, int]]`	The environment to run in.	`None`

Returns:

Type	Description
`Union[Environment, LeanError]`	The resulting environment or an error.

`run_list(commands, env=None)` `async`

Runs a list of commands in the specified environment.

Parameters:

Name	Type	Description	Default
`commands`	`list[str]`	A list of command strings.	required
`env`	`Optional[Union[Environment, int]]`	The Lean environment.	`None`

Returns:

Type	Description
`list[Union[Environment, LeanError]]`	List of results.

`run_file_async(path, return_all_states=False)` `async`

Runs a Lean file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the Lean file.	required
`return_all_states`	`bool`	Whether to return all intermediate states (not fully supported in return type yet).	`False`

Returns:

Type	Description
`Union[Environment, LeanError]`	The resulting environment or an error.

`pickle_env_async(path, env)` `async`

Pickles an environment to a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Destination path.	required
`env`	`Environment`	Environment to pickle.	required

Returns:

Type	Description
`Union[Environment, LeanError]`	Result or error.

`unpickle_env_async(path)` `async`

Unpickles an environment from a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Source path.	required

Returns:

Type	Description
`Union[Environment, LeanError]`	The unpickled environment or error.

`pickle_state_async(path, state)` `async`

Pickles a proof state to a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Destination path.	required
`state`	`ProofState`	Proof state to pickle.	required

Returns:

Type	Description
`Union[ProofState, LeanError]`	Result or error.

`unpickle_state_async(path)` `async`

Unpickles a proof state from a file.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Source path.	required

Returns:

Type	Description
`Union[ProofState, LeanError]`	The unpickled proof state or error.

`close_async()` `async`

Closes the Lean REPL subprocess and its entire process tree.

REPL API Reference

leanflow.SyncREPL

__init__(lean_version=None, repl_path=None, project_path=None, **kwargs)

start()

run(command, env=None)

run_list(commands, env=None)

run_file(path, return_all_states=False)

run_tactic(tactic, state)

close()

leanflow.REPL

__init__(lean_version=None, repl_path=None, project_path=None, timeout=300, header=None, dacite_config=None, fail_on_header_error=True, log_dir=None, log_level=None, debug=False, manage_memory=True, max_memory_mb=8192, max_restart_attempts=3, **kwargs)

load_from_yaml(yaml_path) classmethod

run(command, env=None) async

run_list(commands, env=None) async

run_file_async(path, return_all_states=False) async

pickle_env_async(path, env) async

unpickle_env_async(path) async

pickle_state_async(path, state) async

unpickle_state_async(path) async

close_async() async

`leanflow.SyncREPL`

`init(lean_version=None, repl_path=None, project_path=None, **kwargs)`

`start()`

`run(command, env=None)`

`run_list(commands, env=None)`

`run_file(path, return_all_states=False)`

`run_tactic(tactic, state)`

`close()`

`leanflow.REPL`

`init(lean_version=None, repl_path=None, project_path=None, timeout=300, header=None, dacite_config=None, fail_on_header_error=True, log_dir=None, log_level=None, debug=False, manage_memory=True, max_memory_mb=8192, max_restart_attempts=3, **kwargs)`

`load_from_yaml(yaml_path)` `classmethod`

`run(command, env=None)` `async`

`run_list(commands, env=None)` `async`

`run_file_async(path, return_all_states=False)` `async`

`pickle_env_async(path, env)` `async`

`unpickle_env_async(path)` `async`

`pickle_state_async(path, state)` `async`

`unpickle_state_async(path)` `async`

`close_async()` `async`