Managing a Zoo With Tool Calling and GenAI Using Python, FastAPI, and LangChain

In this tutorial, we are going to show you how you can integrate Langchain into your existing stack even if your endpoints require authenticated users with specific roles.

By the end of this post, you’ll have two running projects built with FastAPI: an API, protected with Auth0 as the application backend, and a small frontend, that will use LangChain for tool calling and will be used by users. The frontend will send user prompts to be processed by LangChain that in turn will invoke the protected API.

The code is available [on this repository](https://github.com/raphaeldovale/auth0-langchain-firstparty) for you to follow along. The repository will have one branch for every significant step in the tutorial: you’ll start on the main branch and move on to `step1` to set up the project.

If you are getting started with tool calling, I recommend you read the blog post linked below before continuing to better understand tool calling in AI agents and how security is currently handled.

> Tool calling intro banner - https://auth0.com/blog/genai-tool-calling-intro/

Zoo AI: A Platform for Managing a Zoo using GenAI

Zoo AI is a zoo management platform that helps staff manage daily operations and respond to various situations. The platform has four main roles:

Veterinarians - Handle animal health and medical care.
Coordinators - Oversee operations and can trigger emergency protocols.
Janitors - Maintain cleanliness and hygiene.
Zookeepers - Provide daily care and feed the animals.

Any staff member can interact with our AI to report when something is happening or when some staff may need to take actions. Also, anything that happens to the animals needs to be logged.

Prerequisites

You will need the following tools and services to build this application:

Python 3.11 or later
Poetry for dependency management. Check how to install.
An Auth for GenAI account. Create one.
An OpenAI account and API key. Create one or use any other LLM provider supported by LangChain.

Zoo AI Technological Stack

The Zoo AI management platform consists of two projects: An API and an Agent. Both projects will have a similar stack to make the platform as simple as possible for demonstration purposes.

Both projects will also use the same Poetry environment to simplify set up. Poetry is completely optional but makes dependency management easier. You can check the Poetry website for more details.

API Project (zoomanagement/api) Stack

FastAPI: Web framework for building the REST API
Auth0-fastapi-api: Authentication and authorization backend with the Auth0 FastAPI SDK
TinyDB: Local NoSQL database for persistence
Pydantic: For data validation and serialization

Agent Project (zoomanagement/agent) Stack

LangChain: Framework for building the AI agent
OpenAI: Language model integration
FastAPI: Web framework for exposing agent endpoints
Requests: Library for HTTP requests
Auth0-fastapi: Authentication and authorization for Web Applications

Getting Started

To begin, clone the repository and install the dependencies. All dependencies needed are already set in pyproject.toml or requirements.txt file (if you are using a different dependency management other than Poetry).

git clone https://github.com/raphaeldovale/auth0-langchain-firstparty

cd auth0-langchain-firstparty

git switch step1

poetry install



Using branch step1 you’ll have an empty project with all dependencies, for both projects, configured.

Creating The Zoo Management API

In this section, you are going to create the Zoo Management API. You are going to use FastAPI and you will learn how to use OAuth2.0 to authenticate users and how to manage dependency injection. By the end of this section, you will have a working API that uses TinyDB as a database.

The API will have four endpoints:

GET /animal: List all animals
POST /animal/{animal_id}/status: Update the status of an animal
GET /staff/notifications: List all notifications for a staff member role
POST /staff/notifications/{role}: Create a new notification for a staff member role

With these endpoints, a staff member can check animal status, update them and also notify other staff members about something that happened.

To start, let’s create the api folder:

mkdir api

cd api



Defining the API Schema

Pydantic is the ideal library to use with FastAPI. It is a data validation and serialization library that also helps FastAPI in validating that the expected data in each endpoint contains what is needed for that endpoint to properly function. It guarantees that you have exactly the data you need, with the rules you defined before.

Create a new file called schema.py and add the below code:

import enum

from datetime import datetime

from pydantic import BaseModel

class UpdateAnimalStatusRequest(BaseModel):

status: str

class NotifyStaffRequest(BaseModel):

description: str

class StaffRole(str, enum.Enum):

JANITOR = "JANITOR"

VETERINARIAN = "VETERINARIAN"

COORDINATOR = "COORDINATOR"

ZOOKEEPER = "ZOOKEEPER"

class AnimalStatus(BaseModel):

time: datetime = datetime.now()

status: str

user_role: StaffRole

user_id: str

class Animal(BaseModel):

id: str

name: str

specie: str

age: int

last_status: list[AnimalStatus]

class StaffNotification(BaseModel):

time: datetime = datetime.now()

description: str

destination_role: StaffRole

notifier_role: StaffRole

notifier_id: str



The above schema will be used for our API and also for the database.

Warning: Ideally in a production code, you should have a different schema for the database and the API and they have different methodologies.

Setting up the database

TinyDB is a lightweight, local, NoSQL database. It is not suited for production, but it is perfect for rapid prototyping as all files are stored in a single JSON file. Create a new file called catalog.py and add the code below:

from tinydb import Query, TinyDB

from schema import Animal, AnimalStatus, StaffNotification, StaffRole

class ItemNotFound(Exception):

pass

class AnimalCatalog:

def __init__(self, db: TinyDB):

self.table = db.table("animals")

self.Animal = Query()

def get_all(self) -> list[Animal]:

return [Animal(**item) for item in self.table.all()]

def add_status(self, animal_id: str, status: AnimalStatus):

result = self.table.search(self.Animal.id == animal_id)

if not result:

raise ItemNotFound

# prepend the new status to the list

result[0]["last_status"].insert(0, status.model_dump(mode="json"))

self.table.update(result[0], self.Animal.id == animal_id)

class StaffNotificationCatalog:

def __init__(self, db: TinyDB):

self.table = db.table("staff_notification")

self.StaffNotification = Query()

def get_notifications_by_role(

self, staff_role: StaffRole

) -> list[StaffNotification]:

notifications_by_role = self.table.search(

self.StaffNotification.destination_role == staff_role.value

)

return [StaffNotification(**item) for item in notifications_by_role]

def add_notification(self, notification: StaffNotification):

self.table.insert(notification.model_dump(mode="json"))



There are two catalogs: one for the animals and another for the notifications. Both have similar structures: both receive a TinyDB instance and create a table for the data.

The actual database is not created on this file, but injected in the constructor for each catalog in another moment: feel free to ignore it for now as this solution will leverage FastAPI’s dependency injection engine, this helps with code maintainability and testability.

The animal table will have one document for each animal. Each document must have the animal ID, name, species, age, and a list of statuses. Each new status will be added to the beginning of the list so the most recent status is the first item on the list.

The notification table will have one document for each notification. Each document has the notification ID, description, destination role, notifier role, and notifier ID. This solution allows us to have a list of notifications for each role but also allows us to know who is the notifier and who is on the receiving end of the notification.

Now we need to create the database itself and load some sample data. Create a new file called db.py and add the code below:

import logging

from pathlib import Path

from typing import Generator

from tinydb import TinyDB

from schema import Animal

logger = logging.getLogger(__name__)

def initialize_db() -> Generator[TinyDB]:

logger.info("Starting database")

Path("./data").mkdir(exist_ok=True)

db_path = "./data/db.json"

file_exists = Path(db_path).exists()

db = TinyDB(db_path)

if not file_exists:

__load_start_data(db)

try:

yield db

finally:

db.close()

def __load_start_data(db: TinyDB):

logger.info("DB does not exist. Loading start data")

zoo_animals = [

Animal(id="ALEX", name="Alex", specie="Lion", age=4, last_status=[]),

Animal(

id="KING_JULIEN", name="King Julien", specie="Lemur", age=12, last_status=[]

Animal(id="MORT", name="Mort", specie="Mouse lemur", age=50, last_status=[]),

Animal(id="SKIPPER", name="Skipper", specie="Penguin", age=35, last_status=[]),

Animal(id="MARTY", name="Marty", specie="Zebra", age=10, last_status=[]),

Animal(

id="GLORIA", name="Gloria", specie="Hippopotamus", age=6, last_status=[]

Animal(id="PRIVATE", name="Private", specie="Penguin", age=10, last_status=[]),

Animal(id="KOWALSKI", name="Kowalski", specie="Lion", age=3, last_status=[]),

]

db.table("animals").insert_multiple([animal.model_dump() for animal in zoo_animals])



Do you know these animals? If you recognize their names you probably also like to move it, move it.

The initialize_db function is a generator that will yield a TinyDB instance. This is a good practice to ensure that the database is closed after use. It also detects if the database file exists and, if not, loads the sample data.

Now, let’s see how this function will be called.

Creating FastAPI Dependencies to be Injected

Since we have the database and both the notification and animal catalogs, you need to create the catalog instances and allow them to be injected into the endpoints.

Create a new file called dependencies.py and add the code below:

from typing import Generator

from fastapi import Depends

from tinydb import TinyDB

from db import initialize_db

from catalog import AnimalCatalog, StaffNotificationCatalog

def get_db() -> Generator[TinyDB]:

yield from initialize_db()

def get_animal_catalog(db: TinyDB = Depends(get_db)) -> AnimalCatalog:

return AnimalCatalog(db)

def get_staff_notification_catalog(db: TinyDB = Depends(get_db)) -> StaffNotificationCatalog:

return StaffNotificationCatalog(db)



The Depends function from FastAPI tells the endpoints that it needs to find a suitable instance of the type we are passing as a parameter to be used. In get_animal_catalog and get_staff_notification_catalog we ask FastAPI to inject the database instance into our function. This is a powerful feature that is used in many ways, including in the authentication process as we are going to see in the coming sections.

Note: All the code until now is on branch step2 of our repository. You can double-check your code or continue from this point.

Setting up authentication in the Zoo API

Now we have the database configured and we can start creating the endpoints. In this section, we will use the Auth0 FastAPI API to make proper authentication.

Before moving forward, we need to set up a few things in the Auth for GenAI account you created in the prerequisites section.

Note: Since Auth for GenAI is in developer preview, you need to use this link and sign up for a new tenant as it will have enable all the features you’ll need.

We need to create a new API. The API will be used to obtain an access token so that the Agent can use it to perform actions on behalf of the user.. Go to the API section in Applications and click on Create API.

We need to set the name and the identifier. The identifier is the audience of the token. We will use it to set the audience of the token in the Agent project. Although the identifier is similar to a URL, it doesn’t have to be. It is just a string that will be used to identify the API.

Name: Zoo Management API
Identifier: https://zoo-management-api

After clicking Create you’ll be taken to the API page. Navigate to the Settings tab check “Allow Offline Access” so we can generate Refresh Tokens for this API.

Now we have to set up roles and inject them into our application access token.

In the Auth0 Dashboard, go to the User Management section and click in Roles.

Now, click on Create Role and set the name COORDINATOR.

Do the same for all remaining roles: janitor, veterinarian, zoo keeper. Keep the same name for both as they will be identified by it.

To finish, we need Auth0 to add the role as a custom claim in the ID and access tokens.

> banner to custom claims blog post - http://auth0.com/blog/adding-custom-claims-to-id-token-with-auth0-actions/

Go to Actions -> Library section. In the Create Action button, select Build from scratch.

Give the action a name (e.g. Add user role) and select the Trigger Login / Post Login

You will be prompted to a code editor screen. Add the following code:

exports.onExecutePostLogin = async (event, api) => {

const namespace = 'https://zooai/roles';

if (event.authorization) {

api.accessToken.setCustomClaim(namespace, event.authorization.roles);

api.idToken.setCustomClaim(namespace, event.authorization.roles);

}

};



Click on Deploy and the code will be ready to be used.

To actually use the Auth0 Action you just created, we need to set up a Trigger for it. Go to Actions -> Triggers and click on the post-login trigger.

Now, just drag and drop the custom action Add user role to the authentication flow and hit Apply.

The action code will then be executed every time a token is generated and will add the claim https://zooai/roles to the access and ID tokens. You should always use a namespace (e.g. https://zooai/) to avoid conflicts with other claims.

Auth0 is configured. Now we need to update our code.

Create a .env file in api folder with the following content:

AUTH0_DOMAIN=YOUR_AUTH0_TENANT_DOMAIN

API_AUDIENCE=https://zoo-management-api



To get AUTH0_DOMAIN you can access any Application, access Settings menu and copy the Domain value.

To finish, create the file auth.py and add the code below.

import os

from dotenv import load_dotenv

from fastapi import Depends, HTTPException, status

from fastapi_plugin import Auth0FastAPI

from schema import StaffRole

load_dotenv()

auth0 = Auth0FastAPI(

domain=os.getenv("AUTH0_DOMAIN"),

audience=os.getenv("API_AUDIENCE"),

)

def require_authenticated_user(claims: dict = Depends(auth0.require_auth())):

return claims

def get_user_role(claims: dict = Depends(require_authenticated_user)) -> StaffRole:

roles = claims.get("https://zooai/roles", [])

if not roles:

raise HTTPException(

status_code=status.HTTP_403_FORBIDDEN,

detail="User has no roles assigned"

)

if len(roles) != 1:

raise HTTPException(

status_code=status.HTTP_403_FORBIDDEN,

detail=f"User must have exactly one role, found: {roles}"

)

try:

return StaffRole[roles[0]]

except KeyError:

raise HTTPException(

status_code=status.HTTP_403_FORBIDDEN,

detail=f"Invalid role: {roles[0]}"

)

def get_user_name(claims: dict = Depends(require_authenticated_user)) -> str:

return claims.get("name", "")

def get_user_id(claims: dict = Depends(require_authenticated_user)) -> str:

return claims.get("sub", "")



If you check the auth.py file you just created, you will see that the get_user_role function is already using the https://zooai/roles claim to get the user role.

Creating the Endpoints and Configuring the Application

Now, create the file main.py and add the below code. This code will have FastAPI initialization and endpoints.

import logging

from fastapi import Depends, FastAPI

from fastapi.responses import JSONResponse

from auth import require_authenticated_user, get_user_id, get_user_role

from dependencies import get_animal_catalog, get_staff_notification_catalog

from catalog import AnimalCatalog, ItemNotFound, StaffNotificationCatalog

from schema import (

Animal,

AnimalStatus,

NotifyStaffRequest,

StaffNotification,

StaffRole,

UpdateAnimalStatusRequest,

)

logger = logging.getLogger(__name__)

app = FastAPI()

@app.exception_handler(ItemNotFound)

async def animal_not_found_exception_handler(request, exception):

return JSONResponse(

status_code=404,

content={"message": "Item not found"},

)

@app.post("/animal/{animal_id}/status")

def update_animal_status(

data: UpdateAnimalStatusRequest,

animal_id: str,

user_claims=Depends(require_authenticated_user),

animal_catalog: AnimalCatalog = Depends(get_animal_catalog),

animal_catalog.add_status(

animal_id,

AnimalStatus(

status=data.status, user_role=get_user_role(user_claims), user_id=get_user_id(user_claims)

)

@app.get("/animal")

def list_animals(

user_claims=Depends(require_authenticated_user),

animal_catalog: AnimalCatalog = Depends(get_animal_catalog),

) -> list[Animal]:

return animal_catalog.get_all()

@app.get("/staff/notification")

def get_staff_notification(

role: StaffRole = Depends(get_user_role),

staff_catalog: StaffNotificationCatalog = Depends(

get_staff_notification_catalog

) -> list[StaffNotification]:

return staff_catalog.get_notifications_by_role(role)

@app.post("/staff/notification/{role}")

def notify_staff(

role: StaffRole,

notification: NotifyStaffRequest,

user_claims=Depends(require_authenticated_user),

staff_catalog: StaffNotificationCatalog = Depends(

get_staff_notification_catalog

logger.info("Storing notification for role %s: %s", role, notification.description)

staff_catalog.add_notification(

StaffNotification(

notifier_role=get_user_role(user_claims),

notifier_id=get_user_id(user_claims),

description=notification.description,

destination_role=role,

)



The code has all endpoints we talked about in the beginning of this post, they all need authentication as we defined Dependency(require_authenticated_user) for each method. We also defined a custom exception handler for the ItemNotFound exception so every time this exception is raised, we return a 404 status code for the client.

Now we can run the API and test it! Into the api folder, run the following command in api folder:

poetry run uvicorn main:app --reload



You can access any endpoint by using curl:

curl http://localhost:8000/animal



Which should give you a “No Authorization provided” error as we did not pass any authentication token. Our Agent will get the token to make these calls, but, to test we did everything right so far you can grab a test access token from the Auth0 Dashboard, go to Applications > API > Zoo Management API and then access the Test tab.

Scroll down to the Sending the token to the API section and copy the cURL there, replace the URL with your current one and you should have something like this (access token omitted):

curl --request GET \

--url http://localhost:8000/animal \

--header 'authorization: Bearer eyJh...



And you should see this result on your terminal:

[{"id":"ALEX","name":"Alex","specie":"Lion","age":4,"last_status":[{"time":"2025-06-22T11:26:51.244102","status":"Alex seems sleepy.","user_role":"COORDINATOR","user_id":"auth0|6856c390a533b4a9021f2013"}]},{"id":"KING_JULIEN","name":"King Julien","specie":"Lemur","age":12,"last_status":[]},{"id":"MORT","name":"Mort","specie":"Mouse lemur","age":50,"last_status":[]},{"id":"SKIPPER","name":"Skipper","specie":"Penguin","age":35,"last_status":[]},{"id":"MARTY","name":"Marty","specie":"Zebra","age":10,"last_status":[]},{"id":"GLORIA","name":"Gloria","specie":"Hippopotamus","age":6,"last_status":[]},{"id":"PRIVATE","name":"Private","specie":"Penguin","age":10,"last_status":[]},{"id":"KOWALSKI","name":"Kowalski","specie":"Lion","age":3,"last_status":[]}]



As we pass a valid access token, you can retrieve all the animals our database has. Not all endpoints will return a valid result, though: the notification endpoints (/staff/notification base path) will return an error as the generated token is not attached to a user and, consequentially, does not have a role.

Now your users should be making API calls like this, that’s what the Agent will do, let’s set it up.

Note: All the code until now is on branch step3 of our repository. You can double-check your code or start at this point.

Creating The Agent Project

Until now, we have created a conventional API that can be used by any client and protected it with Auth0. In this section, we will rely on Langchain and OpenAI to create an agent that can be used to manage the zoo. The project will also use FastAPI and HTML+JS in the frontend to test the agent. We do not need to implement most of the authentication logic as we will rely on the Auth0 FastAPI library and the Universal Login Page provided by Auth0.

The agent project will have only two endpoints:

POST /prompt: Send a prompt to the agent and receive a response
GET /staff_notifications: Get the notifications for the current user role

Setting up the Agent project

We need to create the project folder. In the project root, run the following commands:

mkdir agent

cd agent



The first file we need to create is the agent.py file. This file will have the agent and the integration with the API. Let's split it into some snippets so we can discuss each one of them.

import logging

import os

from dotenv import load_dotenv

import requests

from langchain.agents import AgentExecutor, create_tool_calling_agent

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain_core.tools import tool

from langchain_openai import ChatOpenAI

from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)

load_dotenv()

API_BASE_URL = os.getenv("API_BASE_URL")

def _get_headers(token: str) -> dict:

return {"Authorization": f"Bearer {token}"}



The section above contains the imports and the environment variable we are going to use to set the API base URL. Now we define the tools that translate the API endpoints into Langchain tools.

def list_animals_tool(token: str) -> Callable:

@tool

def list_animals(event: str) -> list[dict]:

"""Get the list of all animals and their IDs."""

logger.info("list animals %s", event)

response = requests.get(

f"{API_BASE_URL}/animal", headers=_get_headers(token=token)

)

response.raise_for_status()

return response.json()

return list_animals



The LangChain Tool is a decorator that indicates that this function can be used by GenAI. You should always add docstrings so the agent can understand what the function does, that’s how the LLM will pick between one tool or the other. To avoid the token to be exposed to the GenAI, we create a function that encapsulates the token and an inner function that will be effectively called by the agent.

class UpdateAnimalStatusArgs(BaseModel):

animal_id: str = Field(..., description="ID of the animal to update")

event_description: str = Field(

..., description="Clear and concise description of the event"

)

def update_animal_status_tool(token: str) -> Callable:

@tool(args_schema=UpdateAnimalStatusArgs)

def update_animal_status(animal_id: str, event_description: str) -> str:

"""Add an event to an animal."""

logger.info("add animal event %s", event_description)

response = requests.post(

f"{API_BASE_URL}/animal/{animal_id}/status",

headers=_get_headers(token=token),

json={"status": event_description},

)

response.raise_for_status()

return "event added"

return update_animal_status



We need to define the arguments that will be passed to the update_animal_status tool. We use Pydantic to define the arguments and the tool decorator to indicate that this function can be used by the LLM. LangChain will automatically parse the arguments and pass them to the function and the configured GenAI will be able to understand each argument.

class NotifyStaffArgs(BaseModel):

event: str = Field(..., description="Event to notify staff about")

staff_role: str = Field(

...,

description="Role of the staff to notify. Can be COORDINATOR, VETERINARIAN, JANITOR or ZOOKEEPER",

)

def notify_staff_tool(token: str) -> Callable:

@tool(args_schema=NotifyStaffArgs)

def notify_staff(event: str, staff_role: str) -> str:

"""Notify a staff group about an event at the zoo."""

logger.info("notify staff %s", event)

response = requests.post(

f"{API_BASE_URL}/staff/notification/{staff_role}",

headers=_get_headers(token=token),

json={"description": event},

)

response.raise_for_status()

return "notification sent"

return notify_staff



The notify_staff tool is similar to the update_animal_status tool. Since we need the exact name values for the role, we instruct the LLM which values are allowed.

def _create_tools(token: str) -> list[Callable]:

return [

list_animals_tool(token),

update_animal_status_tool(token),

notify_staff_tool(token),

]



LangChain expects a list of tools that LLM will be able to call. But, since we need to pass the token to our API, we will create a helper function that holds the token and instantiate the functions for each request.

Now we need to configure the agent and contextualize the LLM. Without the context, the agent will not know how to behave.

# OpenAI model (you need to export OPENAI_API_KEY)

llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")

# OpenAI model (you need to export OPENAI_API_KEY)

llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")

SYSTEM_MESSAGE_PREFIX = """

You are the Smart Zoo AI Assistant. Your primary role is to assist zoo staff with managing operations by intelligently using the available tools.

The user's query will be prefixed with their role and user id (e.g., "Role: 'coordinator'. User ID: 'XXXXX'. \\nQuery: \\n ...").

You MUST pay close attention to the user's role to understand their likely permissions and the context of their request.

Before updating an animal status and notify a staff, check if the report was already made to the animal so we can avoid multiple notifications. If this happens, reply to the notifier that the team is already aware.

Base your understanding of roles on the following:

- COORDINATOR: Can report events, update any animal status, and trigger all actions, including emergency protocols.

- VETERINARIAN: Take care of the animals health

- JANITOR: Take care of the zoo public locations, clean the toilets, etc.

- ZOOKEEPER: Take care of the animals, feed them, clean their cages, etc.

All staff can request emergency actions, this action must be confirmed by a coordinator

Always think step-by-step:

1. Understand the user's intent from their query, username, and role.

2. Consider which tool(s), if any, are appropriate for the request AND the user's role.

3. If the user's role does not permit an action they are requesting, you should state that the action cannot be performed due to role permissions, or suggest an alternative if appropriate.

4. If using a tool, make sure the parameters you provide to the tool are accurate and derived from the user's query.

5. Provide a clear and helpful response to the user.

6. If the user is referring to an animal, locate the animal in the animal database and update its status with the new event.

Also, any notification for the users should pass the animal name, location and any other relevant information

7. Events related to medical attention should be logged in the animal database and notified to the staff.

8. Events related to cleaning attention should be logged in the animal database and notified to the staff.

If you are unsure about an action or if critical information is missing, ask for clarification.

Prioritize safety and adherence to zoo protocols.

Begin!

"""

prompt = ChatPromptTemplate.from_messages(

[

("system", SYSTEM_MESSAGE_PREFIX),

("human", "{input}"),

MessagesPlaceholder(variable_name="agent_scratchpad"),

]

)



The code above sets a prompt that clarifies what is the agent's purpose and how it should behave. We also added some rules so we can set some boundaries to the agent. Feel free to change the prompt and test how the agent changes its behavior.

Now we need to properly create the agent:



async def run_agent(user_input: str, user_role: str, user_id: str, token: str) -> str:

tools = _create_tools(token)

agent = create_tool_calling_agent(llm, tools, prompt=prompt)

agent_executor = AgentExecutor(agent=agent, tools=tools)

message = {

"input": f"Role: '{user_role}'. User id: '{user_id}'. \nQuery: \n{user_input}"

}

logger.info("Query %s", message)

response = await agent_executor.ainvoke(message)

return response["output"]



The function run_agent will be called by our FastAPI endpoint. The endpoint must pass the user prompt (user_input), the user role extracted from the token, the user ID and the access token that will be attached to the requests to the API.

Setting up authentication in the Agent

Now we will work with the authentication configuration. Create the auth.py file and add the following code.

import os

from dotenv import load_dotenv

from fastapi import Depends, Request, HTTPException, status

from auth0_fastapi.auth import AuthClient

from auth0_fastapi.config import Auth0Config

load_dotenv()

auth_config = Auth0Config(

domain=os.getenv("AUTH0_DOMAIN"),

client_id=os.getenv("AUTH0_CLIENT_ID"),

client_secret=os.getenv("AUTH0_CLIENT_SECRET"),

audience=os.getenv("API_AUDIENCE"),

authorization_params={

"scope": "openid profile email offline_access",

"prompt": "consent"

app_base_url=os.getenv("APP_BASE_URL", "http://localhost:3000"),

secret=os.getenv("APP_SECRET_KEY", "SOME_RANDOM_SECRET_KEY"),

mount_connect_routes=True

)

auth_client = AuthClient(auth_config)

def get_user_role (auth_session = Depends(auth_client.require_session)) -> str:

roles = auth_session.user.get("https://zooai/roles", [])

if not roles:

raise HTTPException(

status_code=status.HTTP_403_FORBIDDEN,

detail="User has no roles assigned"

)

if len(roles) != 1:

raise HTTPException(

status_code=status.HTTP_403_FORBIDDEN,

detail=f"User must have exactly one role, found: {roles}"

)

return roles[0]

async def get_access_token(request: Request) -> str:

store_options = {"request": request}

return await auth_client.client.get_access_token(store_options=store_options)



This code relies on auth0_fastapi library to handle all authentication with the user. The library will register all authentication flow endpoints needed and also handle web sessions for you. The session is encrypted using a random key so no one outside your application can read the data.

Now, we need to fill the environment variables used by the code above. With your Auth for GenAI account, you can check the quickstart and simply click on the “Create Application” button. A regular web application will be created with Callback URL point to http://localhost:3000/auth/callback

In your code, create a new .env file and add the following variables:

# Auth0

AUTH0_DOMAIN="YOUR_AUTH0_DOMAIN"

AUTH0_CLIENT_ID="YOUR_AUTH0_CLIENT_ID"

AUTH0_CLIENT_SECRET="YOUR_AUTH0_CLIENT_SECRET"

APP_BASE_URL="http://localhost:3000"

APP_SECRET_KEY="use [openssl rand -hex 32] to generate a 32 bytes value"

# OpenAI

OPENAI_API_KEY="YOUR_OPEN_AI_KEY"

API_AUDIENCE="https://zoo-management-api"

API_BASE_URL="http://localhost:8000"



Now we need to create our main file. This is a bit different from the API as we also need some static files to be served. Create a new file called main.py and add the following code:

import os

import requests

from fastapi import Depends, FastAPI, Request, Response

from fastapi.responses import FileResponse, RedirectResponse

from fastapi.staticfiles import StaticFiles

from pydantic import BaseModel

from starlette.middleware.sessions import SessionMiddleware

from dotenv import load_dotenv

from auth0_fastapi.server.routes import router, register_auth_routes

from agent import run_agent

from auth import auth_config, auth_client, get_access_token

import os

import requests

from fastapi import Depends, FastAPI, Request, Response

from fastapi.responses import FileResponse, RedirectResponse

from fastapi.staticfiles import StaticFiles

from pydantic import BaseModel

from starlette.middleware.sessions import SessionMiddleware

from dotenv import load_dotenv

from auth0_fastapi.server.routes import router, register_auth_routes

from agent import run_agent

from auth import auth_config, auth_client, get_access_token

import logging

load_dotenv()

logging.basicConfig(

level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"

)

app = FastAPI()

### Setting auth0 authentication

app.add_middleware(SessionMiddleware, secret_key=os.getenv("APP_SECRET_KEY"))

app.state.auth_config = auth_config

app.state.auth_client = auth_client

register_auth_routes(router, auth_config)

app.include_router(router)

### Setting up static files

app.mount("/static", StaticFiles(directory="static"), name="static")

### Setting up app routes

@app.get("/")

async def serve_homepage(request: Request, response: Response):

try:

await auth_client.require_session(request, response)

except Exception as e:

logging.error(f"Error requiring session: {e}")

return RedirectResponse(url="/auth/login")

return FileResponse("static/index.html")

class Prompt(BaseModel):

prompt: str

@app.post("/prompt")

async def query_genai(

data: Prompt,

request: Request,

auth_session = Depends(auth_client.require_session),

result = await run_agent(

data.prompt,

user_role=auth_session['user']['https://zooai/roles'\]\[0\],

user_id=auth_session['user']['sub'],

token=await get_access_token(request),

)

return {"response": result}

@app.get("/staff_notifications")

async def get_staff_notifications(request: Request):

access_token = await get_access_token(request)

response = requests.get(

f"{os.getenv('API_BASE_URL')}/staff/notification",

headers={"Authorization": f"Bearer {access_token}"},

)

response.raise_for_status()

return response.json()



The first part of the file handles authentication endpoints. Our endpoints use Depends(auth_client.require_session) to provide proper authentication. If the user hits our application without proper authentication, our library will automatically redirect to the Universal Login page.

Creating pages for the user to interact with

When a user navigates to the root URL, the static/index.html file will be served. This file is a simple HTML file that will be used to display the agent interface, copy the code below into your index.html file.

<!DOCTYPE html>

Zoo AI

Hello! How can I help you today?

</div>

Notifications

</div>



Now create the static/app.js file. This file will send user prompts and check periodically for new notifications for the user.

async function sendMessage() {

const input = document.getElementById('message-input');

const message = input.value.trim();

if (message) {

const messagesContainer = document.getElementById('chat-messages');

messagesContainer.innerHTML += `

${message}

input.value = '';

const response = await fetch(`/prompt`, {

method: 'POST',

headers: {

'Content-Type': 'application/json',

body: JSON.stringify({ prompt: message })

});

const data = await response.json();

messagesContainer.innerHTML += `

${data.response}

}

async function fetchNotifications() {

const response = await fetch(`/staff_notifications`);

const notifications = await response.json();

displayNotifications(notifications);

}

function displayNotifications(notifications) {

const notificationsList = document.getElementById('notifications-list');

notificationsList.innerHTML = notifications.map(notification => `

${notification.notifier_role}

${notification.description}

${new Date(notification.time).toLocaleString()}

</div>

`).join('');

}

document.addEventListener('DOMContentLoaded', async () => {

const messageInput = document.getElementById('message-input');

const sendButton = document.getElementById('send-button');

messageInput.addEventListener('keypress', (e) => {

if (e.key === 'Enter') sendMessage();

});

sendButton.addEventListener('click', sendMessage);

await fetchNotifications();

setInterval(fetchNotifications, 30000);

});



Before running the application, we need to create some users in Auth0, each one with the roles you need. Go to User Management -> Users and create as many users as you want. In the user page, go to the Roles tab and assign the role you desired. Our code only supports one single role per user

Running the Zoo Agent Interface

You can now run the application, go to the agent folder and run the following command (do not forget to keep the API project running).

poetry run uvicorn main:app --reload --port 3000



Note that we are using port 3000 to run the agent application since the API Project is already running on the default port 8000.

Now, go to the browser and open the URL http://localhost:3000/. You should see the application running. On the first time, you will be redirected to the Auth0 login page. After login, you will be redirected to the application and you should see the notifications of the user role.

Your application should be ready to be used. You can login with the user you created in Auth0 and you should see the notifications of the user role. Go play with it and see how it reacts to many different queries!

Note: All the code until now is on branch step4 of our repository.

Recap

In this blog post you learned how to create a FastAPI API with the use of Dependency Injection and Role Based authentication with OAuth and Auth0. To use this API, you learned how to set up and deploy a Langchain application integrated with OpenAI and Auth0. This application enabled the LLM to securely connect to your API with context.

If you are building with GenAI you should check out our brand new product Auth for GenAI, to enable AI agents to securely access tools, workflows, and data with fine-grained control and just a few lines of code.

You can take a look at the code that was used in these demos on GitHub to try it out for yourself and stay tuned because we will have more content on tool calling in Python in future blog posts.