This documentation provides an overview of the code used to implement a RESTful API server for the pipeline using
the SuperagentX library and FastAPI. The server processes incoming requests, authenticates users via API tokens,
and handles search queries related to the trip planner.
Overview
The code sets up a FastAPI server that processes search requests using the AgentXPipe class from SuperagentX.
It incorporates a custom authentication mechanism to validate client requests via an API token. The API exposes
an endpoint that allows users to submit a query and receive results from the trip planner pipeline.
Setup Environment
$ pip install superagentx 'fastapi[standard]'
Implementation
Import the dependencies.
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, HTTPException # https://fastapi.tiangolo.com/
from fastapi.security import APIKeyHeader
from superagentx.agentxpipe import AgentXPipe
from superagentx.result import GoalResult
from trip_planner.config import AUTH_TOKEN
from trip_planner.pipe import get_trip_planner_pipe
The lifespan function initializes the trip planner pipeline when the FastAPI application starts and clears
it when the app shuts down. The pipes dictionary is used to store the pipeline for later use in request handling.
@asynccontextmanager
async def lifespan(app: FastAPI):
pipes['trip_planner_pipe'] = await get_trip_planner_pipe()
yield
pipes.clear()
A custom dependency function, verify_api_token, checks the validity of the API token passed in the api-token header.
If the token is invalid, the function raises an HTTP 401 Unauthorized error.
async def verify_api_token(
api_token: str = Depends(APIKeyHeader(name='api-token', auto_error=False))
):
if api_token != AUTH_TOKEN:
raise HTTPException(status_code=401, detail='Invalid API Token!')
The main API application (ecom_app) is created using FastAPI. The /search endpoint processes search queries
related to the trip planner. The verify_api_token function is included as a dependency to authenticate requests before processing them.
@ecom_app.get('/search', dependencies=[Depends(verify_api_token)])
async def search(query: str) -> list[GoalResult]:
trip_planner_pipe: AgentXPipe = pipes.get('trip_planner_pipe')
return await trip_planner_pipe.flow(query_instruction=query)
This RESTful API provides a simple yet powerful interface for interacting with the trip planner pipeline, ensuring
secure access via API token validation.
Run the script
$ fastapi dev trip_planner/trip_planner/restpipe.py
Result
(venv) ➜ superagentx_examples git:(dev) ✗ fastapi dev trip_planner/trip_planner/restpipe.py
INFO Using path trip_planner/trip_planner/restpipe.py
INFO Resolved absolute path /home/john/Projects/superagentX-examples/superagentx_examples/trip_planner/trip_planner/restpipe.py
INFO Searching for package john structure from directories with __init__.py files
INFO Importing from /home/john/Projects/superagentX-examples/superagentx_examples/trip_planner
╭─ Python package file structure ─╮
│ │
│ 📁 trip_planner │
│ ├── 🐍 __init__.py │
│ └── 🐍 restpipe.py │
│ │
╰─────────────────────────────────╯
INFO Importing module trip_planner.restpipe
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
INFO Found importable FastAPI app
╭─────────── Importable FastAPI app ───────────╮
│ │
│ from trip_planner.restpipe import ecom_app │
│ │
╰──────────────────────────────────────────────╯
INFO Using import string trip_planner.restpipe:ecom_app
╭────────── FastAPI CLI - Development mode ───────────╮
│ │
│ Serving at: http://127.0.0.1:8000 │
│ │
│ API docs: http://127.0.0.1:8000/docs │
│ │
│ Running in development mode, for production use: │
│ │
│ fastapi run │
│ │
╰─────────────────────────────────────────────────────╯
INFO: Will watch for changes in these directories: ['/home/john/Projects/superagentX-examples/superagentx_examples']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [8869] using WatchFiles
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
INFO: Started server process [8882]
INFO: Waiting for application startup.
INFO:watchfiles.main:4 changes detected
INFO: Application startup complete.
INFO:watchfiles.main:4 changes detected