78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
import os
|
|
|
|
import uvicorn
|
|
from dotenv import load_dotenv
|
|
from fastapi import Body, FastAPI
|
|
from google.adk.cli.fast_api import get_fast_api_app
|
|
from google.adk.runners import InMemoryRunner
|
|
from google.genai import types as genai_types
|
|
|
|
from agents.pipeline.agent import root_agent as pipeline_root_agent
|
|
|
|
|
|
def build_pipeline_root(_: str):
|
|
# Agents now live under ./agents and are auto-loaded by ADK FastAPI app.
|
|
# Kept for compatibility with run_sample(); unused in API path.
|
|
from agents.pipeline.agent import root_agent as ra
|
|
return ra
|
|
|
|
|
|
def run_sample():
|
|
load_dotenv()
|
|
|
|
# Allow GEMINI_API_KEY provided in .env to be used by google-genai client
|
|
api_key = os.getenv("GEMINI_API_KEY", "")
|
|
if api_key and not os.getenv("GOOGLE_GENAI_API_KEY"):
|
|
os.environ["GOOGLE_GENAI_API_KEY"] = api_key
|
|
|
|
# Sample disabled: use the FastAPI server instead.
|
|
print("Sample disabled: start the FastAPI app.")
|
|
|
|
|
|
"""FastAPI application configured via ADK's get_fast_api_app, auto-loading agents from ./agents."""
|
|
DEFAULT_ENV = os.path.abspath(os.path.join(os.path.dirname(__file__), ".env"))
|
|
ENV_FILE = os.getenv("ENV_FILE", DEFAULT_ENV)
|
|
load_dotenv(dotenv_path=ENV_FILE)
|
|
if os.getenv("GEMINI_API_KEY") and not os.getenv("GOOGLE_GENAI_API_KEY"):
|
|
os.environ["GOOGLE_GENAI_API_KEY"] = os.getenv("GEMINI_API_KEY")
|
|
|
|
AGENTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "agents")
|
|
SESSION_SERVICE_URI = "sqlite:///./sessions.db"
|
|
ALLOWED_ORIGINS = ["*"]
|
|
SERVE_WEB_INTERFACE = True
|
|
|
|
app: FastAPI = get_fast_api_app(
|
|
agents_dir=AGENTS_DIR,
|
|
session_service_uri=SESSION_SERVICE_URI,
|
|
allow_origins=ALLOWED_ORIGINS,
|
|
web=SERVE_WEB_INTERFACE,
|
|
)
|
|
|
|
_api_runner = InMemoryRunner(agent=pipeline_root_agent)
|
|
|
|
|
|
@app.post("/process_data")
|
|
def process_data(data: dict = Body(...)):
|
|
message = data.get("message") or data.get("text") or data.get("input") or ""
|
|
content = genai_types.Content(parts=[genai_types.Part.from_text(text=message)], role="user")
|
|
|
|
session = _api_runner.session_service.create_session_sync(
|
|
app_name=_api_runner.app_name,
|
|
user_id="api-user",
|
|
)
|
|
|
|
final_text = None
|
|
for event in _api_runner.run(user_id="api-user", session_id=session.id, new_message=content):
|
|
if getattr(event, "content", None):
|
|
for part in event.content.parts or []:
|
|
text = getattr(part, "text", None)
|
|
if text:
|
|
final_text = text
|
|
|
|
return {"emails": final_text or ""}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Run API from this directory: `uv run uvicorn main:app --reload`
|
|
uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 8000)))
|