1  """
  2  FastAPI application module
  3  """
  4  
  5  import inspect
  6  import os
  7  import sys
  8  
  9  from fastapi import APIRouter, Depends, FastAPI
 10  from fastapi_mcp import FastApiMCP
 11  from httpx import AsyncClient, ASGITransport
 12  
 13  from .authorization import Authorization
 14  from .base import API
 15  from .factory import APIFactory
 16  
 17  from ..app import Application
 18  
 19  
 20  def get():
 21      """
 22      Returns global API instance.
 23  
 24      Returns:
 25          API instance
 26      """
 27  
 28      return INSTANCE
 29  
 30  
 31  def create():
 32      """
 33      Creates a FastAPI instance.
 34      """
 35  
 36      # Application dependencies
 37      dependencies = []
 38  
 39      # Default implementation of token authorization
 40      token = os.environ.get("TOKEN")
 41      if token:
 42          dependencies.append(Depends(Authorization(token)))
 43  
 44      # Add custom dependencies
 45      deps = os.environ.get("DEPENDENCIES")
 46      if deps:
 47          for dep in deps.split(","):
 48              # Create and add dependency
 49              dep = APIFactory.get(dep.strip())()
 50              dependencies.append(Depends(dep))
 51  
 52      # Create FastAPI application
 53      return FastAPI(lifespan=lifespan, dependencies=dependencies if dependencies else None)
 54  
 55  
 56  def apirouters():
 57      """
 58      Lists available APIRouters.
 59  
 60      Returns:
 61          {router name: router}
 62      """
 63  
 64      # Get handle to api module
 65      api = sys.modules[".".join(__name__.split(".")[:-1])]
 66  
 67      available = {}
 68      for name, rclass in inspect.getmembers(api, inspect.ismodule):
 69          if hasattr(rclass, "router") and isinstance(rclass.router, APIRouter):
 70              available[name.lower()] = rclass.router
 71  
 72      return available
 73  
 74  
 75  def lifespan(application):
 76      """
 77      FastAPI lifespan event handler.
 78  
 79      Args:
 80          application: FastAPI application to initialize
 81      """
 82  
 83      # pylint: disable=W0603
 84      global INSTANCE
 85  
 86      # Load YAML settings
 87      config = Application.read(os.environ.get("CONFIG"))
 88  
 89      # Instantiate API instance
 90      api = os.environ.get("API_CLASS")
 91      INSTANCE = APIFactory.create(config, api) if api else API(config)
 92  
 93      # Get all known routers
 94      routers = apirouters()
 95  
 96      # Conditionally add routes based on configuration
 97      for name, router in routers.items():
 98          if name in config:
 99              application.include_router(router)
100  
101      # Special case for embeddings clusters
102      if "cluster" in config and "embeddings" not in config:
103          application.include_router(routers["embeddings"])
104  
105      # Special case to add similarity instance for embeddings
106      if "embeddings" in config and "similarity" not in config:
107          application.include_router(routers["similarity"])
108  
109      # Execute extensions if present
110      extensions = os.environ.get("EXTENSIONS")
111      if extensions:
112          for extension in extensions.split(","):
113              # Create instance and execute extension
114              extension = APIFactory.get(extension.strip())()
115              extension(application)
116  
117      # Add Model Context Protocol (MCP) service, if applicable
118      createmcp(application, config)
119  
120      yield
121  
122  
123  def createmcp(application, config):
124      """
125      Create a MCP service if necessary.
126  
127      Args:
128          application: FastAPI Application
129          config: configuration
130      """
131  
132      mcp = config.get("mcp")
133      if mcp:
134          # HTTP Client arguments
135          defaults = {"base_url": "http://apiserver", "timeout": 100}
136          clientargs = mcp.get("clientargs", {}) if isinstance(mcp, dict) else {}
137          clientargs = {**defaults, **clientargs}
138  
139          # Create HTTP client with custom options
140          client = AsyncClient(transport=ASGITransport(app=application, raise_app_exceptions=False), **clientargs)
141  
142          # MCP service arguments
143          mcpargs = mcp.get("mcpargs", {}) if isinstance(mcp, dict) else {}
144  
145          mcp = FastApiMCP(application, http_client=client, **mcpargs)
146          mcp.mount()
147  
148  
149  def start():
150      """
151      Runs application lifespan handler.
152      """
153  
154      list(lifespan(app))
155  
156  
157  # FastAPI instance txtai API instances
158  app, INSTANCE = create(), None