[Enhancement] Read‑Only Production Mode

[mc-dorzo]

Motivation

In production we must guarantee that an agent’s behavior cannot be altered through the public API.

A first‑class read‑only mode gives ops teams an “immutable” deployment switch, preventing any call that could change runtime behavior or stored state.

Solution Proposal

1. Enum for mode:

  class ServerMode(Enum):
      """Server operation modes."""
      DEVELOPMENT = "development"
      PRODUCTION = "production"

3. Create middleware for ProductionMode

class ProductionAccessMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: FastAPI,
        mode: ServerMode,
    ):
        super().__init__(app)
        self.is_prod = mode == ServerMode.PRODUCTION

    async def dispatch(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        path = request.url.path
        method = request.method

        for route in request.app.routes:
            if hasattr(route, "path_regex") and hasattr(route, "methods"):
                if route.path_regex.match(path) and method in route.methods:
                    openapi_extras = getattr(route, "openapi_extra", {})
                    mode = openapi_extras["mode"]

                    if self.is_prod and mode == ServerMode.DEVELOPMENT:
                        raise HTTPException(status_code=404)
                    break

        return await call_next(request)

4. rename apigen_config to api_config function, and add mode like so:

def api_config(
    group_name: str,
    method_name: str,
    mode: ServerMode = ServerMode.PRODUCTION,
) -> Mapping[str, Any]:
    return {
        "openapi_extra": {
            "x-fern-sdk-group-name": group_name,
            "x-fern-sdk-method-name": method_name,
            "mode": mode,
        }
    }

5. add to each endpoint e.g.

e.g.

 @router.post(
        "",
        status_code=status.HTTP_201_CREATED,
        operation_id="create_agent",
        response_model=AgentDTO,
        responses={
            status.HTTP_201_CREATED: {
                "description": "Agent successfully created. Returns the complete agent object including generated ID.",
                "content": example_json_content(agent_example),
            },
            status.HTTP_422_UNPROCESSABLE_ENTITY: {
                "description": "Validation error in request parameters"
            },
        },
        **api_config(
            group_name=API_GROUP,
            method_name="create",
            mode=ServerMode.DEVELOPMENT,
        ),
    )
    async def create_agent(
        params: AgentCreationParamsDTO,
    ) -> AgentDTO: ...

6. add --producton/deployment/read-mode flag

[kichanyurd]

I think --migrate should be allowed to run. I think in many pods, in their startup command --migrate will always be provided in the CLI.

[kichanyurd]

@mc-dorzo Can you provide an estimate for this one please? If it's half a day's work, we should get this out of the way.

[rusty-art]

Could we also consider hardening server.py (and other files) so it doesn't change system path and doesn't depend on current directory.

E.g.

PARLANT_HOME = os.environ.get("PARLANT_HOME")  # e.g. /usr/prod/myproject/parlant
PARLANT_LOGS_DIR = os.environ.get("PARLANT_LOGS") # e.g. /usr/prod/myproject/logs/parlant-logs

# for backwards compatibility - could we delete and raise exception instead?
if not PARLANT_HOME:
        PARLANT_HOME = Path(".").absolute()

PARLANT_DATA_DIR = Path(PARLANT_HOME) / "parlant-data" # replaces PARLANT_HOME_DIR
CONFIG_FILE_PATH = Path(PARLANT_HOME) / "parlant.toml"

if not PARLANT_LOGS_DIR: PARLANT_LOGS_DIR = Path(PARLANT_HOME) / "parlant-logs"
LOGS_FILE_PATH = Path(PARLANT_LOGS_DIR / "parlant.log"

[kichanyurd]

@rusty-art Sounds like an interesting direction, but I'm not sure yet I fully understand.

As far as I'm currently aware, Parlant only changes files under PARLANT_HOME which is controllable by an env variable.

Do you mean that you want to control the location of different files, such that not all of them have to be within the same dir (i.e., PARLANT_HOME)?

And I'm assuming that this is to separate runtime data, logs, and configuration in production environments — for example, to put some of them on read-only mount points, and some of them on writable ones?

Or is there anything else here which I'm missing?

[rusty-art]

Hi @kichanyurd,

Thanks for your reply and questions. Let me clarify the concerns with the current approach and outline recommended best-practices.

Current Issues

1. Mixing Logs and Data
   Parlant uses the PARLANT_HOME directory for both data and log files.

   • This creates confusion: data and log files have different management needs.
   • Data files may be read-only after setup; log files must remain writable and are rotated or deleted regularly.
   • Log files are often needed by central logging systems, which may have different access requirements.
   • Best-practice is to keep logs and data in separate directories (e.g. PARLANT_HOME/data and PARLANT_HOME/logs).

2. Configuration File Location
   The config file path is not built from PARLANT_HOME or an equivalent environment variable.

   • This means config can end up outside the managed directory structure, making deployments and audits harder.
   • Instead, the config file should also sit within the defined structure (e.g. PARLANT_HOME/parlant.toml).

Recommendation

• Treat PARLANT_HOME as the top-level root.
  • Use explicit subdirectories for logs, data, and config.
• Example structure:

PARLANT_HOME/
data/
logs/
parlant.toml

• Avoid using the current working directory or implicit fallbacks.
• If environment variables aren’t set, raise an error rather than defaulting to less secure behaviour.
• If users really want to use current-working-directory they can set environment variable to "."

Summary Table

Purpose	Example Path	Permissions
Data	$PARLANT_HOME/data	Could be read-only (after setup)
Logs	$PARLANT_HOME/logs	Writable, log-rotated
Config	$PARLANT_HOME/parlant.toml	Readable by Parlant

[kichanyurd]

@rusty-art good proposal!

Would you mind creating a separate issue out of it with the suggested changes? We'll do it as part of the Production Hardening milestone.

[mc-kfir]

Hi, Getting back to the original issue:

Motivation

I will start by trying to refine the requirements and suggests some high level solutions before diving in.

1. Let's start with the requirements - what is the purpose of this write-denial ?

• Protection from honest mistakes?
  • End-user who accidentally made a change without being aware
  • Conversation designer who accidentally changed a live production system because they work on dev & prod in parallel
  • Outsource dev which wasn't in sync
• Protection from lightweight malicious actors?
  • "Kids" toying with hacking the system
  • "Prompt pokers" trying to see if they can easily jailbreak the system
  • Competitors trying to embarrass the user (company) in legal ways
• Protection from heavy malicious actors?
  • Black hat hackers
  • Criminal competitors

2. Next questions to ask (answers may be derived from the choice above):

• Protection level: Do we want to only block? also hide? use totally separate sets of interfaces?
• Granularity: Do we want to uniformly block post/put/patch? block by groups? block individual calls?
• Simplicity vs. Dev. Time: Can a minimal solution satisfy the current need ? (easy maintenance & shorter dev time)

Solution ideas (high level)

Some high level options for solutions (for protection against mistakes):

1. Middleware approach : write a slim middleware filter that when a mode is set, will filter requests by their type

• May also filter by endpoint/operation

2. FastAPI Dependency injection: Use this mechanism to insert a configuration checker (checks if this call is allowed by configuration model) to each relevant request handler
3. Function Wrapper: add a wrapper to all relevant API handlers (in router level) that checks the configuration and returns either the func itself or a default/empty func

TBD more details

Discussions

Regarding the flag name: what about --deploy/--deployment ?

[mc-kfir]

Adding here the options for stronger protection:
(It's recommended to read the next comment before this one)

1. Dynamic router filtering - after routers are created and before they are added to app, we can filter the endpoints, as in this example:
   (Note that this code is based on the branch of this PR Test API Infrastructure #364)

@asynccontextmanager
async def configure_agents_router(app: FastAPI, container: Container, deployment_mode: bool) -> AsyncIterator[FastAPI]:
    name="agents"
    router = agents.create_router(
        agent_store=container[AgentStore],
        tag_store=container[TagStore],
    )
    router = filter_router_for_deployment(router, container[APIConfiguration][name], deployment_mode)
    app.include_router(router, prefix=f"/{name}")
    yield app

• Note: Next step will be moving vars "name" & "router" to arg list so it will be easier to change APIConfigurationSteps to labels instead of function (decouple config from code). Leaving this change to a separate PR.

def filter_router_for_deployment(router: APIRouter, allowed_operations: list[str], deployment_mode: bool ) -> APIRouter:
    if deployment_mode:
        return router

    # Filter out design-time routes
    filtered_routes = []
    for route in router.routes:
        if isinstance(route, APIRoute):
            operation_id = route.operation_id
            if operation_id in allowed_operations:
                filtered_routes.append(route)
        else:
            # Keep non-API routes (like sub-routers)
            filtered_routes.append(route)
        
    router.routes = filtered_routes
    
    return router

Notes:

• container[DeploymentConfiguration][router] will probably use a string as key, not the router object itself

2. A variation to this solution is to wait for the full construction of the app, before it starts to serve, then traverse the routers included in it and filter them one by one.
   Pro: The filtering action is done in a central way, in one place
   Con: This is done chronologically and syntactically later - meaning all side effects inside include_router should be undone too and moreover, future side-effects may be added and won't be dealt with (if done in FastAPI for example).
   Also, in such a case where security is a consideration, closing earlier is safer and more controllable. Therefore IMO, the first solution is better.

[mc-kfir]

Command line flag (--deploy/--read-only/--production)

deploy: bool (in case this flag name will be decided) variable will be introduced in StartupParameters, then passed to load_app and in it will be passed to create_api_app and to AppWrapper.
AppWrapper will keep it as a class member which will be used in each configure_ method (configuration steps) called (by the new app configuration scheme).

The configuration:

Structure

A dictionary APIConfiguration: dict[str, list[str]] : Keys are entity names (e.g. "agents", "customers") , values are lists of operation names to allow on deployment (e.g. "read_customer", "list_customers"):

{
    "agents"           : [ "read_agent", "list_agents"],
    "capabilities"     : [ "list_capabilities", "read_capability"], 
    "context-variables": [ "list_variables", "read_variable", "read_variable_value", "update_variable_value"], # delete value ?
    "customers"        : [ "read_customer", "list_customers", "create_customer", "delete_customer" ],
    "evaluations"      : [],
    "guidelines"       : [ "read_guideline", "list_guidelines"] , 
    "index"            : [],
    "journeys"         : [ "list_journeys", "read_journey" ],
    "relationships"    : [ "list_relationships", "read_relationship" ],
    "services"         : [], 
    "sessions"         : [ "create_session", "read_session", "list_sessions", "update_session", "create_event", "list_events", "inspect_event", "delete_events"],
    "tags"             : [ "list_tags", "read_tag" ],
    "terms"            : [ "read_term", "list_terms"],
    "utterances"       : [ "read_utterance", "list_utterances"],
}

Location

The default configuration will be stored in the container (populated in server.py or conftest.py) and used by each configure_XXXX method (defined in app.py)

[mc-kfir]

These changes are the next step of the solution in my earlier comment ("Adding here the options for stronger protection"):

1. Removal of default_configuration_steps (app.py) and assimilation of the information/functionality of APIConfigurationSteps in APIConfiguration which is described in the last comment above.
2. In server.py: when deployment_mode is false then container[APIConfiguration] is assigned None
3. Creation of a function that gets the container, reads the API configuration and returns a dictionary of routers by labels (dict[str, APIRouter] ):

def create_routers(container: Container) -> dict[str, APIRouter]:
    config  = container[APIConfiguration]
    return {
        "agents": agents.create_router(
            agent_store=container[AgentStore],
            tag_store=container[TagStore],
        ) if config is None or config.get("sessions", False) else None,
        "tags": tags.create_router(
            tag_store=container[TagStore],
        ) if config is None config.get("tags", False) else None,
        ...
        glossary.create_legacy_router(
            glossary_store=container[GlossaryStore],
        )
    }

This function will be called in __aenter__ of AppWrapper.
After that, for each non-empty entry of this dictionary, we'll run filter_router_for_deployment (described in a comment above) on it and include it in self.app :

all_routers = create_routers(container)
for entity in all_routers:
    if container[APIConfiguration]:
       router = filter_router_for_deployment(all_routers[entity], container[APIConfiguration][entity])
    else:
       router = all_routers[entity]
    app.include_router(router, prefix=f"/{entity}")

The rest of configure functions (configure_middlewares etc.) will be called as they are called now (as it is in the test-mode branch)