Implementation:Neuml Txtai API Create

Overview

The create() and lifespan() functions in the txtai.api.application module are responsible for constructing the FastAPI application instance, registering security dependencies, loading the YAML-configured txtai application, and dynamically wiring up API routes. Together, they form the complete bootstrap sequence for a txtai API server.

API Signature

create()

from txtai.api.application import create

app = create()

Parameters: None

Returns: A FastAPI application instance with security dependencies registered and the lifespan handler attached.

lifespan(application)

# Called automatically by FastAPI during startup
def lifespan(application):
    ...
    yield

Parameter	Type	Default	Description
application	FastAPI	required	The FastAPI application instance to initialize with routes

Returns: A generator (context manager) -- yields once after startup is complete.

Source Reference

File: src/python/txtai/api/application.py (Lines 31-134)

create() Implementation

def create():
    # Application dependencies
    dependencies = []

    # Default implementation of token authorization
    token = os.environ.get("TOKEN")
    if token:
        dependencies.append(Depends(Authorization(token)))

    # Add custom dependencies
    deps = os.environ.get("DEPENDENCIES")
    if deps:
        for dep in deps.split(","):
            dep = APIFactory.get(dep.strip())()
            dependencies.append(Depends(dep))

    # Create FastAPI application
    return FastAPI(lifespan=lifespan, dependencies=dependencies if dependencies else None)

Key behavior:

• Reads TOKEN environment variable; if set, adds Authorization as a global dependency
• Reads DEPENDENCIES environment variable; if set, resolves and adds each custom dependency class
• Creates the FastAPI instance with the lifespan handler and all collected dependencies
• Dependencies are set to None if the list is empty (no unnecessary overhead)

lifespan() Implementation

def lifespan(application):
    global INSTANCE

    # Load YAML settings
    config = Application.read(os.environ.get("CONFIG"))

    # Instantiate API instance
    api = os.environ.get("API_CLASS")
    INSTANCE = APIFactory.create(config, api) if api else API(config)

    # Get all known routers
    routers = apirouters()

    # Conditionally add routes based on configuration
    for name, router in routers.items():
        if name in config:
            application.include_router(router)

    # Special case for embeddings clusters
    if "cluster" in config and "embeddings" not in config:
        application.include_router(routers["embeddings"])

    # Special case to add similarity instance for embeddings
    if "embeddings" in config and "similarity" not in config:
        application.include_router(routers["similarity"])

    # Execute extensions if present
    extensions = os.environ.get("EXTENSIONS")
    if extensions:
        for extension in extensions.split(","):
            extension = APIFactory.get(extension.strip())()
            extension(application)

    # Add Model Context Protocol (MCP) service, if applicable
    if config.get("mcp"):
        mcp = FastApiMCP(application, http_client=AsyncClient(timeout=100))
        mcp.mount()

    yield

Startup sequence:

1. Configuration loading: Reads CONFIG env var and parses YAML via Application.read()
2. API instantiation: Creates the API instance (or a custom subclass via API_CLASS), which loads all models and indexes
3. Router discovery: Calls apirouters() to find all available APIRouter modules
4. Dynamic route registration: Only includes routers whose names appear as keys in the configuration
5. Special case handling: Adds embeddings router for clusters, similarity router for embeddings
6. Extension execution: Runs any custom extensions specified via EXTENSIONS env var
7. MCP setup: Mounts Model Context Protocol service if configured
8. Yield: Signals that startup is complete and the server can begin accepting requests

apirouters() Implementation

def apirouters():
    api = sys.modules[".".join(__name__.split(".")[:-1])]

    available = {}
    for name, rclass in inspect.getmembers(api, inspect.ismodule):
        if hasattr(rclass, "router") and isinstance(rclass.router, APIRouter):
            available[name.lower()] = rclass.router

    return available

This function uses Python's inspect module to discover all submodules of the txtai.api package that define an APIRouter instance, returning them as a name-to-router dictionary.

Module-Level Objects

# FastAPI instance and txtai API instances
app, INSTANCE = create(), None

Variable	Type	Description
app	FastAPI	The ASGI application object, referenced by uvicorn as txtai.api:app
INSTANCE	API or None	The global txtai API instance, populated during lifespan startup

get() Helper

def get():
    return INSTANCE

Route handlers call application.get() to access the global API instance. This indirection allows the instance to be initialized lazily during the lifespan event.

start() Helper

def start():
    list(lifespan(app))

Used by the AWS Lambda handler to manually trigger the lifespan startup sequence outside of a normal ASGI server context.

Usage Examples

Standard Server Launch

# Set configuration path and start server
CONFIG=config.yml uvicorn "txtai.api:app" --host 0.0.0.0 --port 8000

With Token Authentication

CONFIG=config.yml TOKEN=$(echo -n "secret" | sha256sum | awk '{print $1}') \
    uvicorn "txtai.api:app"

With Custom API Class

CONFIG=config.yml API_CLASS=mypackage.CustomAPI \
    uvicorn "txtai.api:app"

With Extensions

CONFIG=config.yml EXTENSIONS=mypackage.CORSExtension,mypackage.MetricsExtension \
    uvicorn "txtai.api:app"

Environment Variable Reference

Variable	Required	Description
CONFIG	Yes	Path to YAML configuration file
TOKEN	No	SHA-256 hash for bearer token authentication
API_CLASS	No	Fully qualified class name for custom API implementation
DEPENDENCIES	No	Comma-separated list of custom FastAPI dependency classes
EXTENSIONS	No	Comma-separated list of extension classes to execute at startup

See Also

• Neuml_Txtai_API_Server_Bootstrap - Principle behind the ASGI bootstrap architecture
• Neuml_Txtai_Application_Init - How Application.__init__ loads the YAML configuration
• Neuml_Txtai_Authorization_Init - The Authorization dependency registered by create()
• Neuml_Txtai_REST_API_Endpoints - The router endpoints dynamically registered during lifespan

Requires Environment

• Environment:Neuml_Txtai_API_Server_Configuration