Integrating Python and Node Backends with a Frontend

Before Starting

This lab integrates the following projects. For full context, read:

• Spec-Driven Development Approach
• Lab Reviewer API

Purpose and Decisions

A few weeks ago, I built a Agent Backend Python to refine my labs before publishing them on LinkedIn.

In my previous post, I explored Spec-Driven Development by creating an Node Backend. At that point, I had two separate backends that did not communicate with each other:

• a Node.js backend, responsible for authentication (I called it Auth Backend Node);
• a Python(FastAPI) backend, responsible for my lab-review workflow (I called it Agent Backend Python).

That felt like a good opportunity to integrate both services. The Agent Backend Python would receive authentication, and I could evolve the project with a frontend instead of interacting with the API only through Postman or FastAPI docs.

Before implementing anything, I compared a few architectural options. I had implemented similar integrations before, but I wanted to choose the option that best matched this lab's purpose.

Integration Scenarios

Scenario 1: Use the Node Auth Backend Node as the Entry Point

One option was to call the Agent Backend Python from the Node Auth Backend Node.

Frontend
    |
    v
Auth Backend Node
    |
    v
Agent Backend Python

The advantages were clear:

• the frontend would know only one API;
• CORS management would be simpler;
• security rules could be centralized in one place.

The disadvantages were also important:

• the Auth Backend Node would become a gateway or orchestrator;
• it would move away from its original responsibility, which was authentication only;
• every request to the Agent Backend Python would add another network hop.

This approach starts to look like an API Gateway pattern or a Backend for Frontend (BFF). It is a good option if the service is intentionally designed to be a gateway, but it was not the best fit for this lab because I wanted the Auth Backend Node to stay focused.

Scenario 2: Let the Agent Backend Python Validate the Token

Another option was to allow the frontend to send the JWT issued by the Auth Backend Node directly to the Agent Backend Python. By validating and extracting the token claims, the Python service could identify the user associated with the request without requiring the frontend to pass additional user data.

Frontend
    |
    +------> Auth Backend Node
    |
    +------> Agent Backend Python
                 |
                 +------> validate token / extract user claims

Authorization: Bearer <jwt>

In the first version of this idea, the Agent Backend Python could call the Auth Backend Node on every request to validate the token or fetch user details.

The advantages were:

• clear separation of concerns;
• independent services;
• easier evolution toward a microservice architecture.

The disadvantages were:

• extra network overhead if Python called Auth for every request;
• runtime coupling between both services;
• the Auth Backend Node could become a bottleneck.

The pattern behind this option is token-based authentication, similar to the OAuth resource-server model. This was the strongest option, but I wanted to refine it so the Python service would not need to call the Auth Backend Node on every request.

Scenario 3: Use an API Key

Another possible shortcut was to create an API key in the backend, expose it to the frontend, and pass user information through headers.

Frontend
   |
 API Key
   |
Agent Backend Python

I rejected this option quickly.

API keys identify applications, not users. They do not give me real login identity, user sessions, roles, permissions, or user-level authorization. Also, a frontend is a public environment, so placing an API key there would not give me meaningful security.

API keys are better suited for:

• service-to-service communication;
• automation;
• integrations where the client can keep secrets safely.

Scenario 4: Add an API Gateway

An API Gateway was also an appealing option. The frontend would never talk directly to the backends. Instead, it would call the gateway, and the gateway would route each request to the correct service.

        +--------------+
        | API Gateway  |
        +--------------+
            /      \
           /        \
          v          v
 Auth Backend Node   Agent Backend Python

This architecture can centralize cross-cutting concerns such as:

• authentication;
• JWT validation;
• rate limiting;
• logging and monitoring;
• SSL termination;
• request routing.

The main disadvantage was that it would be over-engineering for this stage. I had only two services: an Auth Backend Node and a Python Agent Service. Adding a gateway would mean another component to deploy, configure, monitor, and maintain before it solved a real problem.

This architecture will become more valuable if the system grows. For this lab, it would add complexity too early.

Final Architecture Decision

I chose an improved version of Scenario 2: the Auth Backend Node issues the JWT, and the Agent Backend Python validates the JWT locally.

                +---------------------+
                |  Auth Backend Node  |
                +---------------------+
                        ^
                        |
                      Login
                        |
Frontend ----------------+
    |
    | Authorization: Bearer <jwt>
    v
+-----------------------+
| Agent Backend Python  |
+-----------------------+

The flow is simple:

1. The user logs in.
2. The Auth Backend Node issues a JWT.
3. The frontend stores the token.
4. The frontend calls the Agent Backend Python with Authorization: Bearer <jwt>.
5. The Agent Backend Python validates the JWT.
6. The Agent Backend Python receives an application-level authenticated user.

This was the best option for my scenario because it is:

• simple;
• scalable;
• close to an industry-standard resource-server model;
• aligned with clear separation of concerns.

The frontend still needs to integrate with both backends, but that is acceptable for a small lab. If the project grows, adding an API Gateway later should be straightforward because the frontend API boundaries are already explicit.

Frontend Decision

For the frontend, the decision was simpler.

One option was to build it as a reusable library or treat it as a microfrontend from the beginning. I decided not to do that because it would introduce unnecessary complexity at this stage.

The frontend is currently intended to serve one application. Creating a reusable library now would add maintenance overhead without a clear benefit. If I need to reuse the same frontend across multiple projects or teams in the future, I can extract shared components or adopt a microfrontend architecture later.

For now, the frontend's job is to make the Python Agents backend usable through a real interface instead of Postman or /docs.

Validating JWTs in the Agent Backend Python

The first requirement was to configure the Agent Backend Python with the same JWT secret used by the Auth Backend Node.

For local testing, I used:

JWT_SECRET=local-test

For a real environment, this must be replaced by a secure secret and managed through a proper secret-management strategy.

Why I Used the Strategy Pattern

The Agent Backend Python should not depend permanently on one authentication provider or one token-validation mechanism. It should depend on a contract responsible for verifying tokens.

Today, the concrete implementation validates JWTs locally using a shared secret. In the future, I could replace that with:

• JWKS validation;
• remote token introspection;
• another authentication provider;
• a different token format.

By isolating verification behind a TokenVerifier abstraction, the authentication mechanism becomes interchangeable while the application code stays clean and loosely coupled.

The Strategy Pattern appears in three places:

• core/auth/token_verifier.py defines the contract;
• core/auth/jwt_token_verifier.py implements the current JWT strategy;
• core/auth/dependencies.py selects and uses the configured verifier.

Auth Module Structure

I created a new auth folder in the Agent Backend Python:

core/auth/
  schemas.py
  token_verifier.py
  jwt_token_verifier.py
  dependencies.py

The responsibilities are:

• schemas.py defines AuthenticatedUser with stable application-level fields such as id, email, profile_id, and application_id;
• token_verifier.py defines the TokenVerifier protocol and authentication-related exceptions;
• jwt_token_verifier.py validates signed JWTs and maps claims to AuthenticatedUser;
• dependencies.py exposes FastAPI dependencies such as get_current_user.

Token Verifier Contract

The contract is intentionally small:

from typing import Protocol

from core.auth.schemas import AuthenticatedUser


class TokenVerifier(Protocol):
    def verify(self, token: str) -> AuthenticatedUser:
        ...

The application does not need to know whether the user came from a locally signed JWT, JWKS, introspection, or another provider. It only needs an AuthenticatedUser.

JWT Token Verifier

The concrete implementation validates the token, requires important claims, checks the expected application, and maps token claims to the app user schema.

class JwtTokenVerifier:
    """Verify Auth Backend Node JWTs and map claims to the app user schema."""
    ...

    def verify(self, token: str) -> AuthenticatedUser:
        if not self.secret:
            raise TokenVerifierConfigurationError("AUTH_JWT_SECRET is required.")

        try:
            payload = jwt.decode(
                token,
                self.secret,
                algorithms=[self.algorithm],
                options={"require": list(self.REQUIRED_CLAIMS)},
            )
        except PyJWTError as exc:
            raise TokenVerificationError("Invalid bearer token.") from exc

        return self._build_authenticated_user(payload)

The verify() method receives a JWT and extracts the data from it. Then it calls _build_authenticated_user(), which creates the AuthenticatedUser defined in schemas.py.

FastAPI Dependencies

The dependencies.py file exposes the authentication dependency used by protected routes.

...
async def get_current_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
) -> AuthenticatedUser:
    if credentials is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authorization token required.",
        )

    try:
        return get_token_verifier().verify(credentials.credentials)
    except TokenAuthorizationError as exc:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Token is not authorized for this application.",
        ) from exc
    except TokenVerificationError as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authorization token.",
        ) from exc
    except TokenVerifierConfigurationError as exc:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Token verifier is not configured.",
        ) from exc

get_current_user() receives bearer credentials, validates the token, and returns the authenticated user. The rest of the application receives user data without dealing with JWT internals.

One important detail is that get_token_verifier() returns TokenVerifier, even though it currently instantiates JwtTokenVerifier. Because this is Python, the protocol is structural: any class that satisfies the verify(token: str) -> AuthenticatedUser contract can act as a TokenVerifier.

The current selector is minimal:

def get_token_verifier() -> TokenVerifier:
    if settings.AUTH_TOKEN_VERIFIER != "jwt":
        raise TokenVerifierConfigurationError(
            f"Unsupported AUTH_TOKEN_VERIFIER: {settings.AUTH_TOKEN_VERIFIER}."
        )
    return JwtTokenVerifier()

In the future, the selector could evolve into something like this:

def get_token_verifier() -> TokenVerifier:
    verifiers = {
        "jwt": JwtTokenVerifier,
        "jwks": JwksTokenVerifier,
        "introspection": IntrospectionTokenVerifier,
    }

    verifier_class = verifiers.get(settings.AUTH_TOKEN_VERIFIER)
    if verifier_class is None:
        raise TokenVerifierConfigurationError(
            f"Unsupported AUTH_TOKEN_VERIFIER: {settings.AUTH_TOKEN_VERIFIER}."
        )

    return verifier_class()

Protecting Routes

After that, I protected the routes that require authentication by injecting get_current_user.

router = APIRouter(
    prefix="/labs",
    tags=["Blog Post Writer"],
    dependencies=[Depends(get_current_user)],
)

outputs_router = APIRouter(
    prefix="/outputs",
    tags=["Outputs"],
    dependencies=[Depends(get_current_user)],
)

This means those routes now require a valid bearer token before the request reaches the business logic.

Testing the Authentication Integration

At this point, the local architecture looked like this:

                    ┌───────────────────────┐
                    │     `Auth Backend Node`      │
                    │   localhost:3001      │
                    └───────────▲───────────┘
                                │
                     Login      │ JWT
                                │
┌───────────────────────┐       │
│    Labs Frontend      │───────┘
│   localhost:5173      │
└───────────┬───────────┘
            │
            │ Authorization: Bearer <JWT>
            ▼
┌───────────────────────┐
│  Labs Reviewer API    │
│   localhost:3015      │
└───────────────────────┘

The frontend did not yet have any feature consuming the Labs Reviewer API, so I created a first /me endpoint that returned the authenticated user based on the token. After implementing it, I tested the flow through FastAPI /docs.

GET /me
Authorization: Bearer <jwt>

Frontend Integration

The next step was to create a frontend that could consume the backend resources and use the Labs Review Agent without uploading files through /docs or Postman.

I intentionally kept the frontend scope small because the main goal of this lab was backend integration. The frontend needed to be useful and clear, but deeper UI improvements could wait for future labs.

The initial scope was:

1. Integrate the /me endpoint from the Python application and map the response to a frontend type.
2. Integrate the /labs-review endpoint so the user could submit a file and receive generated outputs.

I had one small modeling question around /me: it belongs to the Labs Reviewer API, but it is part of the authentication structure. For clarity, I placed it in the frontend's auth feature.

Backend Improvements

Once I started adapting the backend for real frontend usage, I realized that the current API needed more than authentication. The user also needed visibility into what was happening during the agent workflow.

Status Endpoint

The first improvement was a status endpoint.

The application had several agents. Some could run independently, while others worked as children of a parent agent. The status response needed to show the current state of each agent clearly.

My first idea looked like this:

{
  "id": "UUID",
  "created_at": "2026-06-09 09:45:00",
  "user_id": "UUID",
  "data": [
    {
      "id": "UUID",
      "name": "Labs Writer",
      "status": "IN_PROGRESS",
      "finished_at": "2026-06-09 09:59:00"
    },
    {
      "id": "UUID",
      "name": "Labs Translator",
      "status": "SUCCEEDED",
      "finished_at": "2026-06-09 09:59:00"
    }
    ...
  ]
}

However, some agents depend on other agents. For example, Labs Reviewer is used to review the Labs Writer output multiple times. I needed to represent sub-agent progress inside the parent agent.

So I changed the shape to support children:

{
  "id": "UUID",
  "created_at": "2026-06-09 09:45:00",
  "user_id": "UUID",
  "data": [
    {
      "id": "UUID",
      "name": "Labs Writer",
      "status": "IN_PROGRESS",
      "finished_at": "2026-06-09 09:59:00",
      "children": [
        {
          "id": "UUID",
          "name": "Labs Reviewer",
          "status": "IN_PROGRESS",
          "loop_from": 1,
          "loop_to": 3,
          "finished_at": "2026-06-09 09:59:00"
        }
      ]
    },
    {
      "id": "UUID",
      "name": "Labs Translator",
      "status": "SUCCEEDED",
      "finished_at": "2026-06-09 09:59:00"
    },
    {
      "id": "UUID",
      "name": "Labs Metadata",
      "status": "FAILED",
      "finished_at": "2026-06-09 09:59:00"
    }
  ]
}

This structure is easier to represent in the frontend because it matches the workflow hierarchy.

Adding MongoDB

To support this status structure, I needed to persist the process state.

I chose MongoDB because this workflow data is document-shaped and does not require relational modeling at this stage. I planned to store the data in two collections:

• process_status;
• process_agents_status.

I also chose Beanie as the ODM because it supports asynchronous operations and has good FastAPI examples:

https://beanie-odm.dev/

I created a MongoDB Atlas cluster and configured the local environment with:

MONGODB_URI=mongodb://localhost:27017
MONGODB_DATABASE=labs_reviewer

Proxy Pattern

At this point, I wanted to update workflow status automatically based on each agent's progress, without manually calling a status-update function inside every agent.

Both the Decorator and Proxy patterns could solve this because both allow behavior to be added without modifying the original implementation. I chose Proxy because my goal was to transparently intercept agent execution and update status before and after selected method calls.

A Decorator could also work, and in Python with FastAPI it can often feel more natural. However, Proxy matched the mental model I wanted for this lab: an intermediary object controlling and monitoring access to the real agent.

"""Proxy wrappers for tracking agent invocations in MongoDB."""
...
class AgentInvocationProxy:
    """Proxy an agent and persist status around selected method calls."""
    ...
    def __getattr__(self, name: str) -> Any:
        attr = getattr(self._agent, name)
        if name not in self._tracked_methods or not callable(attr):
            return attr

        def _tracked_call(*args: Any, **kwargs: Any) -> Any:
            return self._invoke_tracked(attr, *args, **kwargs)

        return _tracked_call

    def _invoke_tracked(self, method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
        self._invocation_count += 1
        loop_from = self._context.loop_from
        loop_to = self._context.loop_to or self._loop_to
        if self._loop_to is not None:
            loop_from = self._invocation_count
            loop_to = self._loop_to

        agent_process_status = self._run_async(
            self._create_agent_process(loop_from, loop_to)
        )
        if self._child_proxy_factory is not None:
            self._child_proxy_factory(agent_process_status)

        try:
            response = method(*args, **kwargs)
        except Exception as exc:
            self._run_async(
                self._mark_agent_process_failed(agent_process_status, str(exc))
            )
            raise

        self._run_async(
            self._mark_agent_process_succeeded(
                agent_process_status,
                self._extract_result(response),
            )
        )
        return response

    def _run_async(self, coroutine: Coroutine[Any, Any, Any]) -> Any:
        if self._async_runner is not None:
            return self._async_runner(coroutine)
        return asyncio.run(coroutine)

    async def _create_agent_process(
        self,
        loop_from: int | None,
        loop_to: int | None,
    ) -> AgentProcessStatus:
        return await self._status_service.create_agent_process(
            process_status_id=self._context.process_status_id,
            parent_agent_process_status_id=(
                self._context.parent_agent_process_status_id
            ),
            name=self._agent_name,
            loop_from=loop_from,
            loop_to=loop_to,
        )

    async def _mark_agent_process_succeeded(
        self,
        agent_process_status: AgentProcessStatus,
        result: str | None,
    ) -> AgentProcessStatus:
        return await self._status_service.mark_agent_process_succeeded(
            agent_process_status=agent_process_status,
            result=result,
        )

    async def _mark_agent_process_failed(
        self,
        agent_process_status: AgentProcessStatus,
        result: str | None,
    ) -> AgentProcessStatus:
        return await self._status_service.mark_agent_process_failed(
            agent_process_status=agent_process_status,
            result=result,
        )
    ...

The application does not call the real agent directly. It calls AgentInvocationProxy.

When a tracked method such as organize_notes() is invoked, Python's __getattr__ intercepts the call. Because the proxy does not implement organize_notes() itself, __getattr__ returns a wrapper function that runs tracking logic before and after calling the real method on the underlying agent.

The key piece is __getattr__. It allows the proxy to transparently intercept method calls while still delegating execution to the actual agent.

Also, note that the private methods _invoke_tracked, _create_agent_process, _mark_agent_process_succeeded, and _mark_agent_process_failed are responsible for calling the status service and persisting the agent's execution status before, during, and after method execution.

Results

After these changes, the /review endpoint returned a process identifier immediately:

{
  "message": "Processing started.",
  "process_id": "f1b1c7c1-a8c3-46d6-b627-c601d7e80021",
  "output_file": "/home/danii/myProjects/labs-reviewer/public/markdown/blog-test_reviewd.md"
}

Then the frontend could call the process status endpoint and receive the current state:

{
  "id": "f1b1c7c1-a8c3-46d6-b627-c601d7e80021",
  "file": "blog-test.md",
  "created_at": "2026-06-10T14:41:43.679000",
  "user_id": "aed1b808-10b5-4512-8989-1bc2b71336d3",
  "data": [
    {
      "id": "8a690097-004c-4312-936a-4ab67119c8c9",
      "name": "Labs Writer",
      "status": "IN_PROGRESS",
      "loop_from": null,
      "loop_to": null,
      "finished_at": null,
      "children": [
        {
          "id": "79c87a1b-4b6a-44b1-93fb-98b17db4a19d",
          "name": "Labs Code Examples",
          "status": "SUCCEEDED",
          "loop_from": 1,
          "loop_to": 1,
          "finished_at": "2026-06-10T14:41:43.850000",
          "children": []
        }
      ]
    }
  ]
}

When the main agent finished, I could also get the status and result of a specific agent:

{
  "id": "8a690097-004c-4312-936a-4ab67119c8c9",
  "name": "Labs Writer",
  "status": "SUCCEEDED",
  "loop_from": null,
  "loop_to": null,
  "finished_at": "2026-06-10T14:42:19.162000",
  "children": [
    {
      "id": "79c87a1b-4b6a-44b1-93fb-98b17db4a19d",
      "name": "Labs Code Examples",
      "status": "SUCCEEDED",
      "loop_from": 1,
      "loop_to": 1,
      "finished_at": "2026-06-10T14:41:43.850000",
      "children": []
    }
  ],
  "result": "# Test File for Flow Evaluation\n\n## Introduction\nThis document serves as a test file designed to evaluate the flow of information and ensure clarity in communication..."
}

This made the workflow much easier to inspect. For each process, the frontend can show the operational status of each agent. For each parent agent, it can also show the status of child agents. I also added an endpoint to get the status and result of one specific agent, because returning every result in the main status endpoint would make the response too large.

Frontend Status Experience

On the frontend, I used a polling strategy. While the main process was IN_PROGRESS, the frontend polled the main status endpoint every 10 seconds.

The goal was to give the user visual feedback while the agents were running. I also used Codex to generate custom SVG icons that matched the statuses and the visual style of the frontend.

Challenges

• Choosing the right pattern was not obvious. I selected the Proxy pattern for status updates, but in Python with FastAPI the Decorator pattern can feel more natural. I was tempted to change the implementation, but my goal was to explore alternatives and study the trade-offs, so I kept Proxy.

• Cypress tests also became expensive in time and tokens. In many iterations, I skipped the test step because the suite took too long. At one point, I left for coffee and came back while it was still processing.

• The Spec-Driven Development workflow also exposed a frontend maintainability problem. My CSS was being placed into one large src/App.css file. I had not specified that styles should be scoped by component, so the generated CSS kept growing in one place. When the file reached 1,365 lines, even small visual fixes became harder than they should be.

• The biggest personal reminder was about AI usage. At first, I felt the application was under control. But when I needed to make personal adjustments, I felt the discomfort of not knowing some parts of the code deeply enough. That is not a good place to be as an engineer. AI is useful, but I still need to understand the code I am responsible for.

Future Improvements

The current code already supports some good next steps:

1. Add a left sidebar with process history.
2. Add a clear download area for generated resources.
3. Improve the frontend layout.
4. Replace FastAPI Background Tasks with a production-ready queue such as Celery + Redis.
5. Let the user submit notes directly instead of uploading a file.
6. MongoDB is tightly coupled, for the next version I must decouple it using Strategy Pattern.

Conclusion

This lab was a study about integrating multiple backends with a frontend. I compared different architectural options, chose JWT-based integration between a Node.js backend and a Python backend, and used design patterns to keep the code organized.

The most important outcome was not only that the services now communicate. The lab also made the Labs Reviewer application more usable: it now has authentication, process tracking, persisted status, and a frontend that can show what the agents are doing.

It also opened several paths for future labs, especially around better frontend structure, production-ready background processing, and richer process-history features.