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
IO ConsoleWebSocket