diff --git a/src/intersystems_pyprod/_parser.py b/src/intersystems_pyprod/_parser.py
index de11d00..e29401c 100644
--- a/src/intersystems_pyprod/_parser.py
+++ b/src/intersystems_pyprod/_parser.py
@@ -9,7 +9,7 @@
 import importlib
 import importlib.util
 from operator import attrgetter
-
+from ._prod_controls import get_prod_status, start_prod, stop_prod, restart_prod
 from ._method_stubs import STUBS
 
 
@@ -854,7 +854,28 @@ def generate_msg_wrappers(tree, script_name, folder_name, iris_package_name, pyt
 
 # _______________________________________________________________________________________________________________________ generate_msg_wrappers #
 
-
+def production_control(args):
+    """
+    Handle production control commands based on the provided arguments.
+    """
+    if args.restart:
+        print("Restarting production...")
+        output = restart_prod(args.timeout, args.force)
+        print(output["message"])
+    elif args.stop:
+        print("Stopping production...")
+        output = stop_prod(args.timeout, args.force)
+        print(output["message"])
+    elif args.start:
+        print(f"Starting production {args.start}...")
+        output = start_prod(args.start)
+        print(output["message"])
+    elif args.status:
+        output = get_prod_status()
+        if output["status"]==2: 
+            print(f"No productions are currently running in {os.environ.get('IRISNAMESPACE', 'the current namespace')}.")
+        else:
+            print(f"Production {output['name']} is {output['status_message']}")
 
 def get_package_root(module_name: str) -> Path:
     spec = importlib.util.find_spec(module_name)
@@ -877,8 +898,22 @@ def main(argv: list[str] = None):
     parser.add_argument("-m", "--module", help="Dotted module to analyze (e.g. pkg.sub.mod). If set, ignore positional file.")
     parser.add_argument("-s", "--source-root", help="Project source root used to compute absolute module names when loading from a file", dest="sourceroot")
     parser.add_argument("input_script", nargs="?", help="Path to a .py file (used when -m/--module is not provided)")
+    
+
+    # Production control options: 
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument("-r", "--restart", action="store_true", help="Restart the production if it's already running")
+    group.add_argument("--start", metavar="PRODUCTION", help="Start the named production")
+    group.add_argument("--stop", action="store_true", help="Stop the production if it's running")
+    group.add_argument("--status", action="store_true", help="Get the status of the production")
+    parser.add_argument("--timeout", type=int, default=10, help="Timeout in seconds for stopping/restarting the production (use with --stop or --restart)")
+    parser.add_argument("--force", action="store_true", help="Force stop the production without waiting for graceful shutdown (use with --stop or --restart)")
+    
+
     args = parser.parse_args(argv)
 
+    prod_control_args = any([args.restart, args.stop, args.start, args.status])
+
     # Load source and module under a correct package context
     if args.sourceroot:
         sys.path.insert(0, os.path.abspath(args.sourceroot))
@@ -909,6 +944,10 @@ def main(argv: list[str] = None):
     else:
         # File mode
         if not args.input_script:
+            # Allow running production controls without providing a script
+            if prod_control_args:
+                production_control(args)
+                sys.exit(0)
             print("You must provide either -m/--module or a path to a .py file.")
             sys.exit(2)
         try:
@@ -1006,7 +1045,8 @@ def main(argv: list[str] = None):
         loaded_module,
         args.module
     )
-
+    if prod_control_args:
+        production_control(args)
 
 
 if __name__ == "__main__":
diff --git a/src/intersystems_pyprod/_prod_controls.py b/src/intersystems_pyprod/_prod_controls.py
new file mode 100644
index 0000000..494ee52
--- /dev/null
+++ b/src/intersystems_pyprod/_prod_controls.py
@@ -0,0 +1,100 @@
+import iris
+
+RUNNING = 1
+STOPPED = 2
+TROUBLED = 3
+SUSPENDED = 4
+
+
+STATUS_MESSAGES = {
+    RUNNING: "Running",
+    STOPPED: "Stopped",
+    TROUBLED: "Troubled",
+    SUSPENDED: "Suspended",
+}
+
+
+class ProductionStatusError(RuntimeError):
+    pass
+
+def get_prod_status():
+    """
+    Returns the name and status of the production as a dictionary.
+    Status codes are: 
+    1 - Running
+    2 - Stopped
+    3 - Troubled
+    4 - Suspended
+    """
+
+    production_name = iris.ref("")
+    production_status = iris.ref("")
+
+    response = iris.Ens.Director.GetProductionStatus(production_name, production_status)
+    if response != 1:
+        raise ProductionStatusError(f"Failed to get production status: {response}")
+    return {"name": production_name.value, "status": production_status.value, "status_message": STATUS_MESSAGES.get(production_status.value, "Unknown")}
+
+
+def start_prod(production_name):
+    """Starts the production."""
+
+    if not production_name or not production_name.strip():
+        return {"ok": False, "message": "Production name is required to start a production."}
+
+    prod = get_prod_status()
+    if prod['status'] in [RUNNING, TROUBLED, SUSPENDED]: # Running, Troubled, or Suspended
+            if prod['name'] == production_name:
+                return {"ok": False, "message": f"Production {prod['name']} is already active with status {prod['status_message']}. Use -r flag to restart the production."}
+            else:
+                return {"ok": False, "message": f"Production {prod['name']} is already active with status {prod['status_message']}. Please stop (--stop) the production before starting a new one."}
+
+    response = iris.Ens.Director.StartProduction(production_name)
+    if response == 1:
+            return {"ok": True, "message": f"Production {production_name} started successfully."}
+    else:
+        return {"ok": False, "message": f"Failed to start production {production_name}: {response}."}
+
+
+
+def stop_prod(timeout=10, force=False):
+    """Stops the production."""
+
+    
+    if timeout < 0:
+        return {"ok": False, "message": "Timeout must be non-negative."}
+
+    prod = get_prod_status()
+    
+    # Production status code 2 is $$$eProductionStateStopped
+    if prod['status'] == STOPPED:
+            return {"ok": False, "message": "There is no production running in the current namespace. Use --start flag to start a production."}
+    
+
+    response = iris.Ens.Director.StopProduction(timeout, force)
+    if response == 1:
+        return {"ok": True, "message": f"Production {prod['name']} stopped successfully."}
+    else:
+        return {"ok": False, "message": f"Failed to stop production {prod['name']}: {response}. Try increasing the timeout or setting force to 1."}
+
+
+
+def restart_prod(timeout=10, force=False):
+    """Restarts the production."""
+    # iris.Ens.Director internally checks if a production is running before restarting it
+    
+    if timeout < 0:
+        return {"ok": False, "message": "Timeout must be non-negative."}
+
+    prod = get_prod_status()
+    if prod['status'] == STOPPED:
+            return {"ok": False, "message": "There is no production running to restart in the current namespace. Use --start flag to start a named production."}
+    
+    response = iris.Ens.Director.RestartProduction(timeout, force)
+
+    if response == 1:
+        return {"ok": True, "message": "Production restarted successfully."}
+    else:   
+        return {"ok": False, "message": f"Failed to restart production: {response}, try increasing the timeout or setting force to 1."}
+
+
diff --git a/tests/test_production_controls.py b/tests/test_production_controls.py
new file mode 100644
index 0000000..fe02255
--- /dev/null
+++ b/tests/test_production_controls.py
@@ -0,0 +1,249 @@
+import os
+import sys
+import time
+from pathlib import Path
+
+import iris
+import pytest
+
+
+PACKAGE_SRC = Path(__file__).resolve().parents[1] / "src"
+if str(PACKAGE_SRC) not in sys.path:
+    sys.path.insert(0, str(PACKAGE_SRC))
+
+from intersystems_pyprod._prod_controls import (  # noqa: E402
+    RUNNING,
+    STATUS_MESSAGES,
+    STOPPED,
+    TROUBLED,
+    get_prod_status,
+    restart_prod,
+    start_prod,
+    stop_prod,
+)
+from intersystems_pyprod._parser import main as parser_main  # noqa: E402
+
+
+PRODUCTION_NAME = "AllPyComponents.Production"
+SETUP_TIMEOUT = 12
+
+
+def _detect_repo_root() -> Path:
+    ws = os.environ.get("GITHUB_WORKSPACE")
+    if ws:
+        return Path(ws).resolve()
+
+    here = Path(__file__).resolve()
+    for path in [here] + list(here.parents):
+        if (path / "pyproject.toml").exists() or (path / ".git").exists():
+            return path
+    return Path.cwd()
+
+
+def _read_director_status():
+    production_name = iris.ref("")
+    production_status = iris.ref("")
+    response = iris.Ens.Director.GetProductionStatus(production_name, production_status)
+    assert response == 1, f"GetProductionStatus failed with status {response}"
+    return {
+        "name": production_name.value,
+        "status": production_status.value,
+        "status_message": STATUS_MESSAGES.get(production_status.value, "Unknown"),
+    }
+
+
+def _wait_for_status(expected_status, timeout=SETUP_TIMEOUT, production_name=None):
+    deadline = time.time() + timeout
+    last_status = None
+
+    while time.time() < deadline:
+        last_status = _read_director_status()
+        if last_status["status"] != expected_status:
+            time.sleep(0.5)
+            continue
+        if production_name is not None and last_status["name"] != production_name:
+            time.sleep(0.5)
+            continue
+        return last_status
+
+    raise AssertionError(
+        f"Timed out waiting for production status {expected_status} "
+        f"for {production_name or 'current production'}; last status was {last_status}"
+    )
+
+
+def _stop_active_production(timeout=SETUP_TIMEOUT):
+    current_status = _read_director_status()
+    if current_status["status"] == STOPPED:
+        return current_status
+
+    if current_status["status"] == TROUBLED and current_status["name"] == PRODUCTION_NAME:
+        return _recover_troubled_production(timeout)
+
+    response = iris.Ens.Director.StopProduction(timeout, False)
+    assert response == 1, f"StopProduction failed during test setup with status {response}"
+
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        current_status = _read_director_status()
+        if current_status["status"] == STOPPED:
+            return current_status
+        time.sleep(0.5)
+
+    response = iris.Ens.Director.StopProduction(timeout, True)
+    assert response == 1, f"Force StopProduction failed during test setup with status {response}"
+
+    try:
+        return _wait_for_status(STOPPED, timeout=timeout)
+    except AssertionError:
+        current_status = _read_director_status()
+        if current_status["status"] == TROUBLED and current_status["name"] == PRODUCTION_NAME:
+            return _recover_troubled_production(timeout)
+        raise
+
+
+def _recover_troubled_production(timeout=SETUP_TIMEOUT):
+    response = iris.Ens.Director.StartProduction(PRODUCTION_NAME)
+    assert response == 1, f"StartProduction recovery failed during test setup with status {response}"
+    _wait_for_status(RUNNING, timeout=timeout, production_name=PRODUCTION_NAME)
+
+    response = iris.Ens.Director.StopProduction(timeout, True)
+    assert response == 1, f"Force StopProduction recovery failed during test setup with status {response}"
+    return _wait_for_status(STOPPED, timeout=timeout)
+
+
+def _start_active_production(timeout=SETUP_TIMEOUT):
+    current_status = _read_director_status()
+    if current_status["status"] == RUNNING and current_status["name"] == PRODUCTION_NAME:
+        return current_status
+
+    response = iris.Ens.Director.StartProduction(PRODUCTION_NAME)
+    assert response == 1, f"StartProduction failed during test setup with status {response}"
+    return _wait_for_status(RUNNING, timeout=timeout, production_name=PRODUCTION_NAME)
+
+
+@pytest.fixture(scope="module", autouse=True)
+def load_production_definition():
+    repo_root = _detect_repo_root()
+    cls_host = repo_root / "tests" / "helpers" / "AllPyComponents" / "Production.cls"
+    py_host = repo_root / "tests" / "helpers" / "AllPyComponents" / "AllPyComponents.py"
+    if not cls_host.exists():
+        raise FileNotFoundError(f"IRIS class file not found: {cls_host}")
+    if not py_host.exists():
+        raise FileNotFoundError(f"Python helper file not found: {py_host}")
+
+    parser_main([str(py_host)])
+    status = iris._SYSTEM.OBJ.Load(str(cls_host), "ck")
+    print("production loading status = ", status)
+    _stop_active_production()
+
+    yield
+
+    _stop_active_production()
+
+
+@pytest.fixture(autouse=True)
+def clean_production_state():
+    _stop_active_production()
+    yield
+    _stop_active_production()
+
+
+def test_get_prod_status_reports_stopped_when_no_production_is_running():
+    result = get_prod_status()
+
+    assert result["status"] == STOPPED
+    assert result["status_message"] == "Stopped"
+
+
+def test_start_production_starts_the_named_production():
+    result = start_prod(PRODUCTION_NAME)
+
+    assert result == {
+        "ok": True,
+        "message": f"Production {PRODUCTION_NAME} started successfully.",
+    }
+
+    status = _wait_for_status(RUNNING, production_name=PRODUCTION_NAME)
+    assert status["name"] == PRODUCTION_NAME
+    assert status["status_message"] == "Running"
+
+
+def test_start_production_rejects_an_already_running_production():
+    _start_active_production()
+
+    duplicate_result = start_prod(PRODUCTION_NAME)
+
+    assert duplicate_result == {
+        "ok": False,
+        "message": (
+            f"Production {PRODUCTION_NAME} is already active with status Running. "
+            "Use -r flag to restart the production."
+        ),
+    }
+
+
+def test_restart_production_restarts_the_running_production():
+    _start_active_production()
+
+    restart_result = restart_prod(timeout=10, force=False)
+
+    assert restart_result == {
+        "ok": True,
+        "message": "Production restarted successfully.",
+    }
+
+    status = _wait_for_status(RUNNING, production_name=PRODUCTION_NAME)
+    assert status["name"] == PRODUCTION_NAME
+
+
+def test_stop_production_stops_the_running_production():
+    _start_active_production()
+
+    stop_result = stop_prod(timeout=10, force=False)
+
+    assert stop_result == {
+        "ok": True,
+        "message": f"Production {PRODUCTION_NAME} stopped successfully.",
+    }
+
+    status = _wait_for_status(STOPPED)
+    assert status["status_message"] == "Stopped"
+
+
+def test_forced_stop_production_stops_the_running_production():
+    _start_active_production()
+
+    stop_result = stop_prod(timeout=10, force=True)
+
+    assert stop_result == {
+        "ok": True,
+        "message": f"Production {PRODUCTION_NAME} stopped successfully.",
+    }
+
+    status = _wait_for_status(STOPPED)
+    assert status["status_message"] == "Stopped"
+
+
+def test_stop_production_reports_no_running_production():
+    result = stop_prod(timeout=10, force=False)
+
+    assert result == {
+        "ok": False,
+        "message": (
+            "There is no production running in the current namespace. Use "
+            "--start flag to start a production."
+        ),
+    }
+
+
+def test_restart_production_reports_no_running_production():
+    result = restart_prod(timeout=10, force=False)
+
+    assert result == {
+        "ok": False,
+        "message": (
+            "There is no production running to restart in the current namespace. "
+            "Use --start flag to start a named production."
+        ),
+    }
\ No newline at end of file