Ryan Fernandes

Posted on Jul 17

Building an Agentic Navigation + Autonomous Execution System

#ai #architecture #agents #softwaredevelopment

Connect with me: LinkedIn · Instagram · GitHub

Hello all, it's nice to have you around.

In this article, I will be discussing the project that I delivered in 2025.
It is an AI agent that enables users to navigate a web mobile enterprise product and perform CRUD operations using natural language.

Two partner companies had attempted but failed to develop the same project before I came along. I will walk through the project in stages. I will explain not only what happened but why, and what alternative paths I could have taken. I will end with changes I would make if I could redesign the project from scratch.

Background: how this project made its way to me
System functionality
High-level architecture
Part 1: The navigation sub-agent (RAG-based screen routing)
Part 2: Autonomous Execution Sub-agent (ReAct + MCP)
- Why ReAct
- Why MCP and not hand-crafted function calling?
How to build an MCP tool?
What I'd do differently (retrospective)
Up next

Background: how this project made its way to me

I began working at my startup in 2025. This was mainly due to my past RAG projects:

A Perplexity clone.
An agent-based text-to-SQL translator.
A custom vision transformer.

This startup had made several attempts to develop their agent. They employed third parties. It always failed because the agent AI development toolkit wasn't ready yet.

I received full ownership almost instantly. My first prototype design was too premature. I was provided with someone else's POC document. I was tasked with completing the design right away without discussing the deadline.

Lesson learned:
If you're getting a project with an unclear scope, discuss the schedule verbally before starting work. Don't wait until after your first draft fails.

Everything became easier after I renegotiated this deadline. I connected with a senior developer within the company. He had no background in AI but was proficient in Java Spring Boot. This helped me to get more information about the product. Here's what I've done and for what reason.

System functionality

The agent can resolve two different types of user intents:

1. Navigation (natural language navigation):
The user asks for things like:

"edit user details"
"go to main dashboard"
"onboard a new device"

The agent responds with a widget. This widget contains a button that leads directly to the required screen.

2. Action (natural language CRUD):
The user asks for things like:

"register a new product 'Wesco gas sensor' with [configuration]"

The agent executes it automatically.

The previous solution had used a text-to-SQL component for executing tasks. This queried the database directly.

The problem with this approach:

Any string input provided by a user and used in constructing a query can be a vulnerability for SQL injection.
The string can also be used in prompt injection. This can make the LLM generate a query it should not generate.

Text-to-SQL components are truly handy for analytics tools with heavy read workloads. They require human approval of the query beforehand.

However, using such an approach for automated and unsanctioned writing in a multi-tenant environment is just wrong.

High-level architecture

The Java Spring backend is right in front of everything.
Its role: It serves as a proxy for authentication.

The FastAPI agent service only serves responses if it receives the request having a service-to-service JWT. This is checked at the JWKS endpoint of the Java Spring backend. Auth for clients does not go through the agent.

Why use a proxy in between? Why not let clients connect straight to FastAPI?

It helps in maintaining one place for checking the session and org tenancy.
It allows the AI layer to be deployed, scaled, or replaced without changing client authentication.

Part 1: The navigation sub-agent (RAG-based screen routing)

Intent classification – how and why semantic routing

Every new query has to go through a classification node. It determines whether it's a navigation intent or an autonomous execution intent.

There are three practical choices:

Classify the intent via LLM: Submit the query to a chat model with instructions like "classify into NAVIGATION or AUTONOMOUS_EXECUTION".
Supervised intent classification: Train a small classifier (logistic regression, tiny transformer fine-tuning, etc.) on annotated examples.
Semantic routing: Calculate the embedding of the input query. Match it to embeddings of reference utterances for each route.

I went with semantic routing. Here is my justification:

An LLM call introduces around 200 to 800 ms latency. It has a per-request cost for a simple yes/no answer. That's overkill.
It also introduces nondeterministic behavior at the very beginning of the request path. An LLM sometimes gets confused by its own classifying instructions.
Classifier training is the correct long-term solution. However, without labeled data at launch time, there is simply nothing to train on.

Why semantic routing fills this void:

It's deterministic.
It requires no training data except a few examples per route.
It takes single-digit milliseconds to complete because all it does is a vector comparison.

Machinery-wise:
Each route (navigation, autonomous_execution) is encoded via a few example utterances. The utterances are encoded just once, during initialization.
At inference time, the query itself is encoded in the same way. The cosine similarity of that encoding to each reference utterance is computed. The route with the maximum (mean) similarity wins. This happens only if it exceeds the minimum required threshold. Otherwise, we consider the query to be ambiguous. I did not implement an ambiguity handler, which is one of the problems I highlight in my retrospective.

from semantic_router import Route, RouteLayer
from semantic_router.encoders import HuggingFaceEncoder

navigation_route = Route(
    name="navigation",
    utterances=[
        "take me to the dashboard",
        "I want to edit my profile",
        "open the onboarding screen",
    ],
)

auto_execution_route = Route(
    name="autonomous_execution",
    utterances=[
        "register a new product",
        "update the user's email",
        "delete this device",
    ],
)

encoder = HuggingFaceEncoder(name="sentence-transformers/all-MiniLM-L6-v2")
router = RouteLayer(encoder=encoder, routes=[navigation_route, auto_execution_route])

decision = router("take me to manage organization page")
# decision.name -> "navigation"

Why cosine similarity?
Cosine similarity and dot product are interchangeable when working with normalized embeddings. The two metrics are indifferent to the magnitude of the vector. They only care about its direction – that is, what it means.
Euclidean distance is more susceptible to non-semantically significant magnitude differences. It's a less suitable default metric for such comparisons, in my opinion.

See the semantic-router documentation for instructions on how to tune the threshold. This should be done explicitly based on a validation dataset.

Why the RAG approach for navigating is useful

The alternatives to RAG in this case:

A massive if/else tree.
A massive keyword-matching system.
A one-shot call to the LLM with the entire list of 70+ screens mentioned every single time.

Keyword matching fails when the user makes his request in an unexpected way. That's precisely why you want to use an LLM – to match paraphrases and synonyms like "edit my info", "update user details", "change my profile".

If you include the list of screens in the prompt every time, it works for 70 screens now, but it does not scale. You burn tokens every time on everything irrelevant. This makes requests longer and increases latency. It also increases the chance that the model gets confused by those screens in the prompt context.

The RAG approach solves both these problems:

Retrieval limits the number of relevant screens to a few before the LLM even sees them.
Generation becomes cheap, quick, and reliable.

Schema design (why Postgres + pgvector over seperate vector DB like chroma, Qdrant, ...)

I chose Postgres with the pgvector plugin rather than a dedicated vector database like Qdrant or Milvus. It is quite a hard decision, so it's important to be clear on it.

For a dedicated vector DB:

Optimized ANN indexes for searching huge vectors (up to tens of millions) may work better.
More powerful search primitives are available out-of-the-box.
Horizontal scalability by design.

For pgvector:

One less service to manage and run.
Consistent transactions between the relational schema and vectors. They reside in the same database and can be updated in one transaction.
The crucial point: the organization already runs Postgres in production. They did not want to add any new operational complexity.

With only ~70 screens, none of the dedicated DB advantages made any difference. This is an amount of data small enough that a scan based on cosine distance would be blazingly fast.

The selection would be different if this were an item search across millions of SKUs. With a limited number of UI screens that grows slowly, pgvector is the safer, cheaper solution.

Two tables, separated so that relational metadata (infrequently changing) and embeddings (recomputable if you change your embedding model) could be managed separately:

CREATE TABLE screens (
    screen_id       SERIAL PRIMARY KEY,
    screen_name     TEXT NOT NULL,
    description     TEXT NOT NULL,
    required_params JSONB,
    web_url         TEXT,
    mobile_path     TEXT
);

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE screen_embeddings (
    vector_id   SERIAL PRIMARY KEY,
    screen_id   INTEGER REFERENCES screens(screen_id),
    description TEXT NOT NULL,
    embedding   VECTOR(384)
);

Below is what the two tables actually look like once populated:

screens holding the relational metadata.
screen_embeddings holding the vectors tied back to it by screen_id.

VECTOR(384) is relevant because it has to be compatible with the dimension of the embedding that your model generates.
Important note: This is a very common reason for a hidden bug. Switching embedding models without changing the vector column causes errors, as a 384 and a 768 dimensional vector cannot be compared.

Data used was stored in a navigation.json seed file:

[
  {
    "screen_name": "profile",
    "description": "User's personal profile page showing account details.",
    "required_parameters": ["user_id", "hostname"],
    "web_url": "https://{hostname}/profile/{user_id}",
    "mobile_path": "/profile"
  }
]

{hostname} and {user_id} are important placeholders. This particular product is white-labeled; each client organization will have their own hostname. Screen URLs need to be generated per request, not during ingest.

Why use a template string?
Given 70 URLs and a limited number of placeholders, substitution using .format() is easy to implement, test, and debug. A more structured solution would be required if the logic behind the placeholders was more complex.

Generation embedding – choice of model & why normalization is important

The all-MiniLM-L6-v2 model (384-dimensional embeddings) from sentence-transformers was used.

In terms of architecture, the model space can be divided into:

Small & fast sentence embedding models (MiniLMs, 384 dimensions): Produce good semantic quality on short texts. Run on CPU without problems. Encode in a fraction of a second.
Larger & more accurate models (e.g., bge-large, text-embedding-3-large from OpenAI): Better semantic understanding of longer or ambiguous texts. Take longer to execute. Are paid per use when using a hosted API. sentence transformer

Descriptions of screen elements are short and clear, generated by us. That is exactly the kind of data where a small model will give good semantic quality. Running it locally will save time and money.

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

def embed_text(text: str) -> list[float]:
    return model.encode(text, normalize_embeddings=True).tolist()

Setting normalize_embeddings=True is not merely a stylistic preference. It is necessary to make cosine similarity work as expected when using the <=> operator from the pgvector extension.

Why is this important?
This process rescales all vectors to unit length. It ensures that the comparison depends only on direction/meaning. It does not depend on how "large" the descriptions are. This is a frequent but subtle mistake since retrieval "works well enough," but sorting deteriorates.

The reason I eventually switched away from the local model, due to latency

Using the all-MiniLM-L6-v2 model locally proved to be a significant part of the 5-second latency during navigation. The encoding of one query took about 500ms. This is quite a lot for one forward pass through a fairly small model. This does not seem to be an intrinsic property of the model. It is a consequence of running it on standard application-level infrastructure using CPU, without any batching or GPU.

Rather than tuning the inference pipeline, I moved the embeddings extraction to Amazon Titan Text Embeddings via the Bedrock API. The configured dimension of the embeddings is either 512 or 1024.

This introduces an extra network hop for the Bedrock API compared to local compute. At first glance, it looks like it will make the whole thing slower. However, a managed embedding service provisioned to serve requests ended up faster than an unoptimized local model.

The takeaway: "Local" does not necessarily mean "fast" if your local environment is not sufficiently tuned. It's better to measure the actual numbers.

This was not a plug-and-play replacement. It entailed re-embedding all the rows in screen_embeddings. I also had to convert the VECTOR(n) column to conform to the Titan vector size. A 384-dimensional vector from MiniLM and a 512- or 1024-dimensional vector from Titan cannot be compared.

For the generation step (the LLM call), I chose MiniMax 2.5. The embedding model should be good at generating comparable vectors. The generation model has to be good at instruction-following and structured tool-calling.

Ingestion script example

import json
import psycopg2

conn = psycopg2.connect(dsn=DATABASE_URL)
cur = conn.cursor()

with open("navigation.json") as f:
    screens = json.load(f)

for screen in screens:
    cur.execute(
        """
        INSERT INTO screens (screen_name, description, required_params, web_url, mobile_path)
        VALUES (%s, %s, %s, %s, %s) RETURNING screen_id
        """,
        (screen["screen_name"], screen["description"],
         json.dumps(screen["required_parameters"]),
         screen["web_url"], screen["mobile_path"]),
    )
    screen_id = cur.fetchone()[0]

    vector = embed_text(screen["description"])
    cur.execute(
        """
        INSERT INTO screen_embeddings (screen_id, description, embedding)
        VALUES (%s, %s, %s)
        """,
        (screen_id, screen["description"], vector),
    )

conn.commit()

I ensured that each description was concise enough to fit as a standalone unit. This is significant. Chunking is a technique used when dealing with documents longer than the context size of a model. Chunking a screen description that is already one or two sentences long will serve no purpose. It will only break down the meaning across several vectors, making retrieval more complex.

Screen: Edit User Details

This screen allows administrators to view and modify a user's profile information,
including full name, email address, phone number, assigned role, account status,
and department. The page also provides options to save changes, reset the user's
password, or deactivate the account.

Retrieval and parameter resolution

During querying, the steps are:

Embed the user query.
Find the top-k closest screens based on cosine similarity.
Pass the metadata associated with them to a generation prompt.
Resolve any dynamic path parameters ({hostname}, {org_id}, {user_id}).

def retrieve_screens(query: str, k: int = 3):
    query_vec = embed_text(query)
    cur.execute(
        """
        SELECT s.screen_name, s.web_url, s.mobile_path, s.required_params
        FROM screen_embeddings e
        JOIN screens s ON s.screen_id = e.screen_id
        ORDER BY e.embedding <=> %s
        LIMIT %s
        """,
        (query_vec, k),
    )
    return cur.fetchall()

def resolve_params(url_template: str, context: dict) -> str:
    return url_template.format(**context)

Why k=3 and not k=1?
When we retrieve the only nearest screen, a poor embedding retrieval will silently mislead the user. There is nothing we can do about it. Retrieving the top 3 screens gives the LLM the opportunity to mitigate retrieval errors. It can distinguish the meaning of "edit my details" between "personal profile" and "account settings" screens using finer prompts.

Why two steps instead of one?
When split into two steps, the retrieval step becomes cheaper and cacheable regardless of generation. This way we can easily update any of these components, the embeddings, or the prompts.

The navigation sub-agent is stateless on purpose. It does not require any conversational context to answer "take me to X" requests. Statelessness is a design choice. Introducing memory would only introduce unnecessary overhead and potential bugs for an operation that never requires memory.

Connection Pooling, and Why Raw Connection Doesn't Survive Production Requests

This ingestion script creates one instance of psycopg2.connect() for a one-off batch job. This is fine for a script run once and exit.

This approach would be unacceptable for the retrieval code path. There, the code is executed with each request from each concurrently active user. Creating a new TCP connection to the database on each request is relatively expensive (TCP handshake, TLS handshake, spinning up a new backend Postgres process).

The solution: a connection pool.
A connection pool is a set of already established connections that can be borrowed by requests and reused. I used psycopg_pool for this. I configured the connection pool with min/max limits so the service cannot create more connections than Postgres is configured to accept.

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    conninfo=DATABASE_URL,
    min_size=4,       # kept warm even when idle
    max_size=20,       # hard ceiling, must stay under Postgres's max_connections
    max_idle=300,      # close connections idle longer than 5 minutes
    max_lifetime=1800, # recycle every connection after 30 minutes
    timeout=5,         # how long a request will wait for a free connection
)

def retrieve_screens(query: str, k: int = 3):
    query_vec = embed_text(query)
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT s.screen_name, s.web_url, s.mobile_path, s.required_params
                FROM screen_embeddings e
                JOIN screens s ON s.screen_id = e.screen_id
                ORDER BY e.embedding <=> %s
                LIMIT %s
                """,
                (query_vec, k),
            )
            return cur.fetchall()

The with pool.connection() as conn: construct is doing something besides mere syntactic sugar.

On completion, it gives the connection back to the pool, rather than closing it.
If the block fails due to an exception, the pool determines if that connection should be returned or destroyed.

This is what separates pooling from simply having a globally-accessible connection object. There is no mechanism within a global connection to protect one request's broken transaction from another request. Pooling allows you to replace a connection that ends up in an invalid state.

Handling expiring and stale connections specifically

Database connections that live for a long time do not remain valid indefinitely. This is due to factors beyond your control:

Idle timeout kills on the server side: Many managed PostgreSQL providers will quietly terminate connections left idle beyond a certain threshold.
Network drops: Load balancers, NAT gateways, and cloud networking layers tend to drop idle TCP connections.
PostgreSQL restarts or failovers: A failover of a managed database will invalidate all current connections.

Such a pool will give you a perfectly valid-looking connection that will fail as soon as any query is executed on it.

The proactive part of the solution:

max_lifetime ensures that each connection must expire and be re-established within a certain period. No connection can get so old that it becomes stale on the server side.
max_idle expires connections that have been sitting idle for too long. This circumvents most idle-timeout expirations.

The reactive part of the solution:
A health check during checkout. This detects any connection that has become stale between uses.

from psycopg_pool import ConnectionPool
from psycopg import OperationalError

def check_connection(conn):
    conn.execute("SELECT 1")

pool = ConnectionPool(
    conninfo=DATABASE_URL,
    min_size=4,
    max_size=20,
    max_idle=300,
    max_lifetime=1800,
    check=check_connection,   # run before a connection is handed out
)

This went together with a thin retry wrapper around the execution of the query. It covers the situation where the connection fails during a transaction.

def retrieve_screens_with_retry(query: str, k: int = 3, attempts: int = 2):
    for attempt in range(attempts):
        try:
            return retrieve_screens(query, k)
        except OperationalError:
            if attempt == attempts - 1:
                raise
            # connection died mid-use; loop and let the pool hand out a fresh one
            continue

Summary of what enabled this solution to withstand production traffic:

A bounded pool size to protect Postgres.
max_lifetime / max_idle parameters for rotating connections proactively.
A health check at checkout.
An extremely specific retry strategy on OperationalError.

Output shape:

{
  "navigation": {
    "screen_name": "manage_organization",
    "url": "https://client.example.com/org/482"
  }
}

Latency: caching and why the 0.8 latency in particular

The first p50 latency for navigation was about 5 seconds per query. This is far too long for something so fundamentally a redirect.

Those 5 seconds consisted of:

The embedding of the query.
The round trip from the database to retrieve it.
The dominant latency: the call to the LLM for generation of the reply.

Navigation queries are frequently duplicates of questions asked before.

Embedding-based response caching was added.
If the embedding of a new query was similar to a cached query by a cosine similarity greater than 0.8, I reused the cached result. The parameter resolution was always done fresh.

Why 0.8?
0.8 is a threshold balancing precision against recall, specific to this particular task.

Below 0.8 (e.g., 0.6): Queries will be semantically distinct but still have the same answer cached. This causes potential errors.
Above 0.8 (e.g., 0.95): The cache will almost never return anything, as paraphrasing will rarely produce semantically identical embeddings.

It was determined by testing some examples. This is why my retrospective calls for a proper evaluation on a labeled dataset.

Clarification on the caching process:
The key cannot simply be the raw query itself. That would be useless. It also cannot ignore the individual user information. The same query string from two different users will have two different URLs. What is cached is the result of the retrieval and generation. Parameter resolution is always done fresh.

Part 2: Autonomous Execution Sub-agent (ReAct + MCP)

This was the more challenging half of the system. It was the reason why the attempts by the previous vendors failed.

Why ReAct

Autonomous execution is totally different from navigation. It doesn't involve "finding the one correct solution out of a given set of choices." It involves figuring out how to achieve a certain goal.

"Register a new product Wescco gas sensor with [config]" may involve several steps. Examples include finding the manufacturer id first, then the category id, and finally registering the product. All of these were not stated by the user.

ReAct (Reasoning + Acting) is a pattern where the model alternates between reasoning and tool use. It observes the outcome of each tool before deciding the next step.

Example of reasoning: "I need the manufacturer ID before I can create the product."
Example of tool use: Looking up the manufacturer ID.

A one-shot prompt that plans the entire sequence up front fails when the output of a step determines the next step (e.g., the manufacturer does not exist yet and must be created first). The loop-until-done architecture of ReAct solves this problem naturally.

The real trade-off: Latency and expense. ReAct makes multiple LLM invocations per task instead of just one. This is why autonomous execution latency becomes variable and, on average, much longer than navigation.

Why MCP and not hand-crafted function calling?

All major LLM vendors provide the ability to perform "function calling" or "tool calling" directly from the prompt. You could do this by hand-crafting your own Python functions.

Model Context Protocol (MCP) standardizes the transport & discovery layer. Tools can be called over stdio or HTTP at run-time.

Why was this relevant practically?
The tools used to do tasks were shared infrastructure. They could theoretically be used by other internal tools/agents. The tools required running as an independently deployable and scalable piece of infrastructure. MCP does this for free. The tool server doesn't care which agent is hitting it, as long as it follows the protocol.

How to build an MCP tool?

Building an MCP tool with FastMCP is remarkably straightforward. It feels just like writing standard Python functions. By wrapping your code in the @mcp.tool decorator, FastMCP handles the complex underlying protocol. It automatically generates JSON schemas based on your Python type hints and docstrings. It manages the JSON-RPC communication transport.

You just write the logic and define the inputs. The framework instantly exposes your tools to any MCP-compatible LLM or agent.

from fastmcp import FastMCP

# Initialize the MCP server
mcp = FastMCP("CalculatorServer")

# Use the decorator to expose this function as an MCP tool
@mcp.tool
def calculate_sum(a: int, b: int) -> int:
    """Adds two integer numbers together. This description tells the LLM when to use it."""
    return a + b

if __name__ == "__main__":
    # Starts the server using the standard input/output transport (default)
    mcp.run()

If you want a deeper visual walkthrough, check out Building Python MCP Servers with FastMCP. This video provides a comprehensive guide on building, testing, and connecting MCP servers using the framework.

What ctx actually is, and how to use it

ctx is the context object FastMCP injects into your tool or middleware function. It's your handle into everything about the current call that isn't a declared argument. This includes:

Request headers.
State set by earlier middleware.
Session data.
Utilities like logging or progress reporting.

Full reference here: gofastmcp.com/servers/context.

In this project, ctx is where two crucial things lived:

The user's bearer token, read off the request headers.
The user_email decoded from the JWT by the auth middleware.

Both live on ctx rather than being passed as regular tool parameters. They are set by code, not by the model.

Why each and every one of our tools wraps an existing REST endpoint, and not raw SQL

This is the straightforward solution to the problem of generating SQL code from natural language. Each of our MCP tools wraps an already-authorized REST endpoint at the Java Spring backend.

The implication: The LLM cannot initiate anything other than what the same user could have done using the existing UI. It is limited precisely by the same backend validation and authorization checks. The LLM does not get to create a query. It only provides parameters for some predefined, reviewed action.

from fastmcp import FastMCP

mcp = FastMCP("custom-tools")

@mcp.tool()
async def create_user(name: str, email: str, org_id: str, ctx) -> dict:
    """Register a new user in the given organization."""
    token = ctx.request_context.headers.get("Authorization")
    response = await http_client.post(
        f"{JAVA_BACKEND_URL}/api/users",
        json={"name": name, "email": email, "org_id": org_id},
        headers={"Authorization": token},
    )
    return response.json()

The ctx parameter is where the caller's bearer token comes from. See what ctx actually is above.

Reasons for the collapsing of tools created from chained API calls into a single tool

After reverse engineering the internal API calls, I understood sequences like: manufacturer look-up, then user-id look-up, then creation. In each case, the chain of API calls was collapsed into a single MCP tool. They were not separate tools for each API call.

Justification:
Every extra tool added to the agent's set of tools adds to the reasoning load on the model. It now has to reason about the order of calls and how to pass outputs as inputs. Reducing a known sequence of calls down to one tool removes this sequencing logic. It is done via deterministic Python code, which will always be correct. It leaves the LLM with the simple decision of which tool to call and what parameters to use.

In general: If it's deterministic and you can calculate it at build time, then do so. Don't let the LLM figure it out again on every request.

Where the underlying APIs provided for paging and good default values, I let that pass through unaltered. This lowers the number of arguments the LLM has to juggle. It lowers the probability of an ill-formed API call.

A backup for wrongly classified queries: navigating within the ReAct loop

There are cases where the semantic router doesn't quite get it right. A navigational request gets classified as an API call. This gets sent to the ReAct agent, which has no way to fulfill "take me to the dashboard" using create_user or update_form.

Instead of making the classifier perfect, I put in a safety net on the execution part. I added a navigation tool to the list of tools provided to the ReAct agent. This consists of the exact retrieval and resolution mechanism used for navigation, embedded within a callable function.

from langchain_core.tools import tool

@tool
async def find_screen(query: str, hostname: str, org_id: str, user_id: str) -> dict:
    """Use this when the user's request is actually about navigating to a
    screen in the product, rather than creating, updating, or deleting data."""
    screens = retrieve_screens(query, k=3)
    top = screens[0]
    url = resolve_params(top["web_url"], {"hostname": hostname, "org_id": org_id, "user_id": user_id})
    return {"screen_name": top["screen_name"], "url": url}

react_tools = [*mcp_tools, find_screen]

Two things to clarify:

This tool lives inside FastAPI, not the MCP server. MCP is worth its overhead only when a tool needs to be independently deployable or shareable.
The docstring is important. It is the only clue the ReAct agent has regarding when to use it. It must explicitly indicate the situation.

This addresses only one side of the problem. When a real autonomous execution query is mistakenly directed to the navigation sub-agent, there's no safety mechanism. The navigation sub-agent's RAG pipeline doesn't have any means of executing the write request. This is one of the reasons why the below retrospective calls attention to the problem of confidence-based fallback.

Transient token propagation - the harder bit, explained in depth

Typically, MCP servers work with long-lived credentials. You may keep one token per client in your database and use it many times. In our case, tokens expired in 2 hours. Each incoming request from a user could be performed with another token.

How this breaks the naive approach:
An agent, according to LangChain, should establish a connection to its MCP server once per agent initialization. That means all tool objects are initialized using some credentials available at that time. To use transient per-user tokens naively, you would need to create a new agent object for each incoming request.

The solution:
Separate the moment of binding a tool from supplying the credentials of the tool. The MCP adapter in LangChain allows for tool interceptors. These are call-time callbacks that allow the insertion of request-specific information (like a bearer token) into a tool that was bound at startup time.

from langchain_mcp_adapters.client import MultiServerMCPClient

async def token_interceptor(request, user_token: str):
    request.headers["Authorization"] = f"Bearer {user_token}"
    return request

client = MultiServerMCPClient(
    {
        "nimbly": {
            "transport": "streamable_http",
            "url": "http://localhost:9001/mcp",
        }
    }
)
tools = await client.get_tools(interceptor=lambda req: token_interceptor(req, current_user_token))

On the MCP server-side, the middleware takes the token out of the request ctx on a per-call basis. It then uses it to authenticate the downstream API request.

Wherever an API had to know who was behind the change (audit purposes), I took the email claim from the JWT itself. I did not request the LLM to provide the email as a parameter. This is the trust boundary. The JWT claim is signed and belongs to an authorized session. The tool parameter provided by the LLM is arbitrary text.

@mcp.middleware()
async def auth_middleware(ctx, call_next):
    token = ctx.request.headers.get("Authorization")
    ctx.state["user_email"] = decode_jwt(token).get("email")
    return await call_next(ctx)

Why use streamable HTTP and why co-locate first

Two MCP transports are relevant here:

stdio: Used by tools operating as subprocesses of the agent.
Streamable HTTP: Tools that can be found over the network in a tool server form.

Streamable HTTP was the best choice. This tool server had to be independently addressable. stdio is tightly coupled to a particular agent process lifecycle.

The initial setup had both the MCP server and the FastAPI agent running inside the same container. They were listening on localhost. This was to avoid the additional latency of an extra hop while the system was still relatively small. The two components were eventually separated into individually deployable entities when there was sufficient demand.

The basic idea is to colocate based on latency and simplicity. Only separate them when there is a real, measurable requirement.

Auth chain end to end, and why each hop verifies independently

In this scenario, two separate tokens are involved. Mixing them up would be erroneous.

The service JWT: Used to authenticate Java Spring itself whenever it sends a request to FastAPI. It shows "this request is genuinely sent by our backend." FastAPI verifies this using Java Spring's JWKS endpoint.
The user token: Authenticates the end user for whom the specific write downstream is done. FastAPI does not validate this token. It gets passed through the interceptor directly to the MCP server. It is ultimately authenticated by the downstream Java REST API.

This is by design. FastAPI does not need to know how to validate a user session. Validating the same thing again in another place could lead to inconsistency.

Conversation memory (what a checkpoint actually is)

The agent state, in LangGraph, tracks the user query and a typed final-answer object:

from pydantic import BaseModel
from typing import Optional

class NavigationResult(BaseModel):
    screen_name: str
    url: str

class AutoExecutionResult(BaseModel):
    summary: str
    tool_calls: list[str]

class FinalAnswer(BaseModel):
    navigation: Optional[NavigationResult] = None
    auto_execution: Optional[AutoExecutionResult] = None

class AgentState(BaseModel):
    user_query: str
    final_answer: FinalAnswer = FinalAnswer()

final_answer is a struct that contains both fields. Only the relevant one is set for a particular turn. This was done deliberately. A client interpreting the answer does not have to do a case analysis. They can always look at final_answer.navigation or final_answer.autonomous_execution. The other field remains None.

A new UUID thread_id is assigned to each chat thread. A checkpointer saves the whole state of the graph at the end of each step. It is identified by thread_id. Next time input comes for the same thread_id, LangGraph will restore the last checkpoint state and continue. Hence, the agent will have a memory between steps.

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    result = await graph.ainvoke(
        {"user_query": query},
        config={"configurable": {"thread_id": thread_id}},
    )

Docs: LangGraph persistence

Why use Postgres for the checkpointer?
It comes down to durability and consistency with the rest of the system. An in-memory checkpointer loses all conversation information whenever a process is restarted. It is unusable for production.

A point of confusion worth clarifying:
A checkpoint snapshot includes all knowledge the graph accumulated. This includes reasoning steps and sub-agent calls. This is necessary to resume correctly. However, this is not what should be included in the "history" displayed to the user.

I included a separate chat_messages table. This was specifically logged by the application code when a turn had completed. It is used for rendering clean, de-duplicated conversation history in the UI. The checkpointer's job is to save the working memory of the agent. chat_messages saves the user's view of that memory.

Streaming – Why Per-Node Events Rather Than Waiting for the Entire Response

Responses are streamed from FastAPI to the Java layer using Server Sent Events (SSE). LangGraph's astream creates an event each time a node completes its execution. This is as opposed to the client waiting for the entire multi-step ReAct process to complete.

async def stream_agent_response(query: str, thread_id: str):
    async for event in graph.astream(
        {"user_query": query},
        config={"configurable": {"thread_id": thread_id}},
        stream_mode="values",
    ):
        yield f"data: {json.dumps(event)}\n\n"

This is especially important for task processing. The latency there is unpredictable. Without streaming, the interface would be sitting on a loading screen with no indication of what is being processed. Streaming individual node-level events (e.g., "checking manufacturer") makes it possible for the frontend to see progress.

SSE is useful because the data stream goes in one direction only (server to client) for a single request-response interaction. It is the simpler protocol that does not require a connection to go through HTTP.

Request/response shape

{
  "user_query": "register a new product called Wescco gas sensor",
  "thread_id": "b3f1...",
  "hostname": "client.example.com",
  "platform": "web",
  "metadata": { "user_id": "u_123", "org_id": "org_42" },
  "header": { "user_token": "..." }
}

The org_id variable is important. The tool serves several organizations with entirely separate datasets. Every query and tool function has an implicit scope based on the org_id. This scoping is what prevents the multi-tenant architecture from exposing one organization's data to another. This scoping must take place at the query/function call level, not just in the prompt.

What I'd do differently (retrospective)

Shipping software that works does not equal shipping software that works well. There are a few areas where I would change my approach in retrospect:

No official eval set: The similarity thresholds and the 0.8 cache hit threshold were tuned empirically through spot checks. They were not tested against an annotated set of queries. This is an easy problem to solve with even a small gold set of a few hundred annotated query/route pairs.
Absence of a fail-safe alternative in case of low confidence: When the semantic router score just exceeds the threshold, the design follows the same route. There is no safeguard for an execution query mistakenly routed to navigation.
Token expiration isn't done gracefully mid-job: A ReAct job spanning longer than a 2-hour token lifetime requires a refresh-and-retry pattern. Currently, a job crossing this line will simply fail halfway with a partially completed transaction.
Insufficient auditing of tool calls: Every tool invocation should potentially be included in a structured audit log. I developed the code for extracting emails but not yet aggregated in an audit log.
Did not take into account the expenses of designing on the same day: In itself, not a technical mistake, but an organizational one that caused me more stress. When you get an unclear, critical scope on day one, discuss the time frame before writing, not after your first attempt fails.

Up next

In the next post, I'll cover adding LangSmith for tracing, evaluation, and catching regressions in the classifier and retrieval thresholds before they hit production.

Thanks for reading. If you have any doubts, suggestions, or want to go deeper, I am happy to answer.