Multi Agent workflow with Fastapi

[ZakisCodes]

Hi everyone,

I'm currently exploring Google's Agent Developer Kit (ADK) and trying to integrate it with FastAPI to build a modular agentic system. While the Google ADK Web UI works fine for me, I get stuck every time I try to bring it into a FastAPI backend.

As a beginner in AI agents and multi-agent systems, I often find myself confused—especially about how to structure my code and manage agent lifecycles inside FastAPI routes. I’ve searched around, but haven’t found solid beginner-level resources or working examples that use FastAPI with Google ADK.

Here's my current folder structure for reference:

src/
├── agents/
│   └── editor_agent/
│       ├── sub_agents/
│       │   ├── sub_agent1/
│       │   │   ├── agent.py
│       │   │   └── __init__.py
│       │   ├── sub_agent2/
│       │   │   ├── agent.py
│       │   │   └── __init__.py
│       ├── agent.py
│       └── __init__.py
├── adk_routers.py
├── main.py
├── models.py
└── __init__.py

What I Need Help With:

• Where and how should I initialize the agents in FastAPI?
• How should I structure API endpoints to call agents and return their output?
• Any best practices or working examples using FastAPI with Google ADK?
• Advice for setting up modular sub-agents (as shown above) and triggering them properly via endpoints.

If anyone here has experience with this or can help me lay the foundation for integrating FastAPI and ADK, I’d really appreciate your guidance 🙏. Even basic tips or example repo links would be super helpful.

Thanks in advance!

[jhammarstedt]

We are doing similar stuff, right now we have gotten quite far by just using the endpoints you get from wrapping your application in the from google.adk.cli.fast_api import get_fast_api_app (there are some drawbacks with this of course but should provide you with a good starting point). You're also free to add more routers and endpoints to this. But you get all sessions endpoints and run_sse (agent entrypoint) from there :)

[jhammarstedt]

If you just start by having your different apps in a root folder

src/
├── agents/
│   └── editor_agent/
│       ├── sub_agents/
│       │   ├── sub_agent1/
│       │   │   ├── agent.py
│       │   │   └── __init__.py
│       │   ├── sub_agent2/
│       │   │   ├── agent.py
│       │   │   └── __init__.py
│       ├── agent.py
│       └── __init__.py
├── main.py
├── models.py
└── __init__.py

Then you can just get started with something like:

@asynccontextmanager
async def lifespan(app: FastAPI):
    await startup_event()
    yield
    await shutdown_event()


# Setting this to None so that we don't start up the database session service
# in test mode
db_url = (
    None if get_settings().ENV == "test" else get_settings().SQLALCHEMY_DATABASE_URI
)

app = get_fast_api_app(
    agents_dir="agents",
    web=False, #set to true if you want to use their interface but keep it false to test your own endpoints
    session_db_url=db_url,
    lifespan=lifespan, # here you set your regular lifespan events 
)
app.include_router(my_own_router_1)
app.include_router(my_own_router_2)

[chanchalchitwan]

@jhammarstedt could you plz shed some lights on the drawbacks with this approach?

[ZakisCodes]

That part has been resolved for now, but I'm currently facing a tricky issue with a database connection error. I've managed to work through several other problems, but I'm stuck on this one and could use some help.

The Problem:
I keep getting this error when trying to set up my database connection:

greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/20/xd2s)

My Setup:
I'm using DatabaseSessionService with a Neon Cloud Database (PostgreSQL). Here's what my code looks like:

DB_URL = "postgres+asyncpg://user:password@your-neon-hostname.neon.tech/neondb"

# Create a lifespan event to initialize and clean up the session service
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup code
    print("Application starting up...")
   
    # Initialize the DatabaseSessionService instance and store it in app.state
    try:
        app.state.session_service = DatabaseSessionService(
            db_url=DB_URL,
            connect_args={"ssl": "require"}
        )
        print("Database session service initialized successfully.")
    except Exception as e:
        print("Database session service initialization failed.")
        print(e)
    
    yield 
    
    # Shutdown code
    print("Application shutting down...")
    
app = FastAPI(title="ADK Multi-Agent FastAPI Demo", lifespan=lifespan)

What I've Tried:
I already installed sqlalchemy[asyncio], but the error is still showing up.

Has anyone encountered this before? It seems like there's something wrong with how I'm handling the async database initialization, but I'm not sure what the best approach is. Any suggestions or guidance would be really appreciated!

Thanks in advance for any help! 🙏

[wailbentafat]

the probleeme in teh conection args change it

[runa]

Thanks, I'm evaluating options too: why you're not using adk api_server?

[tahreemrasul]

I have the same question. Although when I use that, I get an internal server error when trying to run the /run endpoint

[dougdew64]

I'm trying but so far failing to understand how to integrate my agents into a web app. It seems that I must deploy my agents to an environment such as the Agent Engine of Vertex AI, and then have my web app call those agents. Is that correct? If so, how do I make the bi-di voice stuff work?

[dougdew64]

https://google.github.io/adk-docs/streaming/custom-streaming/