Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
1593ca0
feat: add FastAPI modular server example with SSE streaming support
FrigaZzz Sep 15, 2025
67e6bf5
Merge branch 'google:main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 15, 2025
7beab8c
refactor: replace __import__("time") with time module import
FrigaZzz Sep 15, 2025
0a3a0a4
refactor: improve code readability and maintainability
FrigaZzz Sep 15, 2025
235bca7
refactor: improve type hints and code organization
FrigaZzz Sep 15, 2025
c393957
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 15, 2025
d4cc10c
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 16, 2025
a7ea7f3
style: reorganize imports for better readability and consistency (./a…
FrigaZzz Sep 16, 2025
c5b3545
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 17, 2025
ff07aae
docs: Update README to reflect changes in custom server file structur…
FrigaZzz Sep 17, 2025
3f88d35
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 18, 2025
9d1f87b
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 18, 2025
a3532e9
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 18, 2025
41284fe
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 19, 2025
ab74d1b
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 19, 2025
4a90e70
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 20, 2025
29e7a19
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 25, 2025
b3e1e78
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 25, 2025
2a0b124
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 26, 2025
5cba2aa
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 27, 2025
f187108
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 29, 2025
5abac2a
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Sep 29, 2025
76f71d6
Merge branch 'main' into feat/fastapi-modular-server-contribution
FrigaZzz Oct 2, 2025
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions contributing/samples/fastapi_modular_server/.env.example
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
# Application Configuration
DEBUG=true
PORT=8881
HOST=localhost
LOG_LEVEL=INFO
LOGGING=true

# ADK Configuration
SERVE_WEB_INTERFACE=true
RELOAD_AGENTS=true

# Google Gemini Configuration
GOOGLE_API_KEY=YOUR_GOOGLE_API_KEY_HERE

# Model Configuration
MODEL_PROVIDER=google
20 changes: 20 additions & 0 deletions contributing/samples/fastapi_modular_server/.gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
# Logs
*.log

# Environment variables
.env

# Python bytecode
*.pyc
__pycache__/

# Test artifacts
.pytest_cache/

# Virtual environments
.venv/
venv/

# IDE configuration
.vscode/
.idea/
260 changes: 260 additions & 0 deletions contributing/samples/fastapi_modular_server/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,260 @@
# πŸš€ Google ADK FastAPI Modular Server

A **production-ready template** for extending Google's Agent Development Kit (ADK) with custom FastAPI endpoints, optimized SSE streaming, and modular architecture patterns.

## 🎯 Purpose & Value

The **FastAPI Modular Server** serves as a **template and reference implementation** for teams who want to:

- **Extend ADK's built-in server** without modifying core behavior
- **Accelerate production deployment** with battle-tested patterns
- **Add custom business logic** through modular router systems
- **Enable hot-reload capabilities** for faster development cycles

## ✨ Key Features

### πŸ”§ **Modular Router Architecture**
- Clean separation of concerns with dedicated router classes
- Easy to add new endpoints without touching core server code

### ⚑ **Optimized SSE Streaming**
- **3 optimization levels** for different use cases:
- `MINIMAL`: Essential content only (author + text)
- `BALANCED`: Core data with invocation tracking
- `FULL_COMPAT`: Complete ADK event compatibility
- Reduced payload sizes for improved performance
- Custom event filtering and mapping

### πŸ”„ **Hot-Reload Development**
- Automatic agent reloading on file changes
- File system monitoring with `watchdog`
- Development-friendly with production stability


## πŸ“ Project Structure

```
fastapi_modular_server/
β”œβ”€β”€ .env.example # Environment variables template
β”œβ”€β”€ README.md # Project documentation
β”œβ”€β”€ __init__.py # Package initialization
β”œβ”€β”€ app/ # Main application directory
β”‚ β”œβ”€β”€ __init__.py # App package initialization
β”‚ β”œβ”€β”€ agents/ # Agent definitions
β”‚ β”‚ └── greetings_agent/ # Greetings agent module
β”‚ β”‚ β”œβ”€β”€ __init__.py # Agent package init
β”‚ β”‚ └── greetings_agent.py # Greetings agent implementation
β”‚ β”œβ”€β”€ api/ # API layer
β”‚ β”‚ β”œβ”€β”€ __init__.py # API package init
β”‚ β”‚ β”œβ”€β”€ routers/ # API route definitions
β”‚ β”‚ β”‚ β”œβ”€β”€ __init__.py # Routers package init
β”‚ β”‚ β”‚ └── agent_router.py # Agent-related API routes
β”‚ β”‚ └── custom_adk_server.py # FastAPI server configuration
β”‚ β”œβ”€β”€ config/ # Configuration management
β”‚ β”‚ └── settings.py # Application settings
β”‚ β”œβ”€β”€ core/ # Core application components
β”‚ β”‚ β”œβ”€β”€ __init__.py # Core package init
β”‚ β”‚ β”œβ”€β”€ dependencies.py # Dependency injection
β”‚ β”‚ β”œβ”€β”€ logging.py # Logging configuration
β”‚ β”‚ └── mapping/ # Data mapping utilities
β”‚ β”‚ β”œβ”€β”€ __init__.py # Mapping package init
β”‚ β”‚ └── sse_event_mapper.py # Server-Sent Events mapper
β”‚ └── models/ # Data models
β”‚ β”œβ”€β”€ __init__.py # Models package init
β”‚ └── streaming_request.py # Streaming data models
└── main.py # Application entry point
```

## πŸš€ Quick Start

### 1. **Configuration**
```bash
# Copy environment template
cp .env.example .env

# Edit .env with your settings
vim .env

# Set the API KEY

```

### 2. **Run the Server**
```bash
# Development mode with hot-reload
python main.py

# Production mode
uvicorn main:app --host 0.0.0.0 --port 8881
```


## πŸ”§ Customization Guide

### **Adding New Routers**

Create a new router following the established pattern:

```python
# app/api/routers/my_custom_router.py
from fastapi import APIRouter, Depends
from app.core.dependencies import ADKServices, get_adk_services

class MyCustomRouter:
def __init__(self, web_server_instance):
self.web_server = web_server_instance
self.router = APIRouter(prefix="/custom", tags=["Custom"])
self._setup_routes()

def _setup_routes(self):
@self.router.get("/endpoint")
async def my_endpoint(
):
# Access any ADK service
sessions = await self.web_server.session_service.list_sessions()
return {"data": "custom response", "session_count": len(sessions)}

def get_router(self) -> APIRouter:
return self.router
```

Register it in the custom server:

```python
# In app/api/custom_adk_server.py - CustomAdkWebServer class
def _initialize_routers(self):
try:
self.agent_router = AgentRouter(self)
self.my_custom_router = MyCustomRouter(self) # Add this
logger.info("All routers initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize routers: {e}", exc_info=True)

def _register_modular_routers(self, app: FastAPI):
# ... existing code ...

if self.my_custom_router:
app.include_router(self.my_custom_router.get_router())
logger.info("Registered MyCustomRouter.")
```

### **Overriding ADK Endpoints**

#### **Method 1: Route Removal (Current Approach)**

```python
def _register_modular_routers(self, app: FastAPI):
# Remove specific ADK routes
routes_to_remove = []
for route in app.routes:
if route.path in [
"/run_sse",
# You could add additional ADK routes here if you want to override them,
# e.g., "/apps/{app_name}/users/{user_id}/sessions"
] and hasattr(route, 'methods') and 'POST' in route.methods:
routes_to_remove.append(route)

# Remove the routes
for route in routes_to_remove:
app.routes.remove(route)
```

#### **Method 2: Middleware Interception**

For more complex overrides, use middleware:

```python
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class RouteOverrideMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Intercept specific routes
if request.url.path == "/run_sse" and request.method == "POST":
# Handle with custom logic
return await self.handle_custom_sse(request)

return await call_next(request)
```

### **Accessing ADK Services and Runners**

#### **From Router Classes**
```python
class AgentRouter:
def __init__(self, web_server_instance):
self.web_server = web_server_instance

async def my_endpoint(self, adk_services: ADKServices = Depends(get_adk_services)):
# Access services
agents = self.web_server.agent_loader.list_agents()
session = await self.web_server.session_service.list_sessions()

# Access runners through web server
runner = await self.web_server.get_runner_async("your_app_name")

# Access other web server properties
runners_cache = self.web_server.runners_to_clean
```

### **Optimizing SSE Streaming**

#### **Custom Event Filtering**

Extend the SSE mapper for more sophisticated filtering:

```python
# app/models/streaming_request.py
class OptimizationLevel(str, Enum):
"""Enumeration for the available SSE optimization levels."""

MINIMAL = "minimal"
BALANCED = "balanced"
FULL_COMPAT = "full_compat"
ULTRA_MINIMAL = "ultra_minimal"

# app/core/mapping/sse_mapper.py
class AdvancedSSEEventMapper(SSEEventMapper):
def map_event_to_sse_message(self, event: Event, optimization_level: OptimizationLevel) -> Optional[str]:
# Custom filtering logic
if self._should_skip_event(event):
return None

# Custom payload creation
payload = self._create_custom_payload(event, optimization_level)

# Custom serialization
return self._serialize_payload(payload)

def _should_skip_event(self, event: Event) -> bool:
# Skip system events, debug events, empty events, etc.
if event.author in ["system", "debug"]:
return True
if not event.content or not event.content.parts:
return True
return False

def _create_custom_payload(self, event: Event, level: OptimizationLevel) -> Dict:
if level == OptimizationLevel.ULTRA_MINIMAL:
# Even more minimal than minimal
return {"t": self._extract_text_only(event)}

return super()._create_minimal_payload(event)
```

## 🀝 Contributing

This template is designed to be extended and customized for your specific needs. Key extension points:

1. **Router Classes**: Add domain-specific endpoints
2. **SSE Mappers**: Custom event processing and optimization
3. **Middleware**: Cross-cutting concerns
4. **Services**: Additional business logic services
5. **Configuration**: Environment-specific settings

## πŸ“š Further Resources

- **Google ADK Documentation**: https://google.github.io/adk-docs/
- **FastAPI Documentation**: https://fastapi.tiangolo.com/
- **Pydantic Settings**: https://docs.pydantic.dev/latest/concepts/pydantic_settings/
- **Server-Sent Events**: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
Empty file.
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from app.agents.greetings_agent.greetings_agent import root_agent
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
from google.adk.agents import LlmAgent

root_agent = LlmAgent(
model="gemini-2.5-flash",
name="greetings_agent",
description="A friendly Google Gemini-powered agent",
instruction="You are a helpful AI assistant powered by Google Gemini.",
tools=[],
)
Empty file.
Loading