This example demonstrates how to build a production-ready API using Flama. It integrates SQLAlchemy for database persistence via DDD (Domain-Driven Design) patterns, defines ML model resources, sets up an MCP (Model Context Protocol) server for tool-calling, and configures an LLM endpoint.
Key architectural components include:
- SQLAlchemy Integration: Uses
SQLAlchemyModule for database connectivity and SQLAlchemyTableRepository for data access. - DDD Layers: Implements a
Worker pattern (SQLAlchemyWorker) to manage repository lifecycles and a Resource layer to handle HTTP routing. - ML/LLM Resources: Deploys a custom ML model via
MLResource and a native LLM model. - MCP Server: Exposes tools, resources, and prompts to an LLM via
MCPServer.
import hashlib
import json
import typing
import sqlalchemy
from pydantic import BaseModel as PydanticModel
import flama
from flama import Flama, types
from flama.ddd.components import WorkerComponent
from flama.ddd.exceptions import NotFoundError
from flama.ddd.repositories.sqlalchemy import SQLAlchemyTableRepository
from flama.ddd.workers.sqlalchemy import SQLAlchemyWorker
from flama.exceptions import HTTPException
from flama.http import APIResponse
from flama.mcp.server import MCPServer
from flama.models import MLResource
from flama.resources import Resource
from flama.resources.routing import ResourceRoute
from flama.sqlalchemy import SQLAlchemyModule, metadata
# --- Database ---
user_table = sqlalchemy.Table(
"user", metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),
sqlalchemy.Column("name", sqlalchemy.String, nullable=False),
sqlalchemy.Column("email", sqlalchemy.String, unique=True, nullable=False),
sqlalchemy.Column("password", sqlalchemy.String, nullable=False),
)
# --- Schemas ---
class UserCreate(PydanticModel):
name: str
email: str
password: str
class UserOutput(PydanticModel):
id: int
name: str
email: str
# --- DDD layers ---
class UserRepository(SQLAlchemyTableRepository):
_table = user_table
class UserWorker(SQLAlchemyWorker):
user: UserRepository
class UserResource(Resource):
name = "user"
verbose_name = "User"
@ResourceRoute.method("/", methods=["POST"], name="create")
async def create(
self,
worker: UserWorker,
data: typing.Annotated[types.Schema, types.SchemaMetadata(UserCreate)],
):
"""
tags:
- User
summary:
Register a new user.
responses:
201:
description: User created.
"""
async with worker:
hashed = hashlib.sha256(data["password"].encode()).hexdigest()
result = await worker.user.create({**data, "password": hashed})
return APIResponse(result[0], status_code=201)
@ResourceRoute.method("/{user_id:int}/", methods=["GET"], name="retrieve")
async def retrieve(
self, worker: UserWorker, user_id: int,
) -> typing.Annotated[types.Schema, types.SchemaMetadata(UserOutput)]:
"""
tags:
- User
summary:
Get user by ID.
"""
async with worker:
try:
return await worker.user.retrieve(id=user_id)
except NotFoundError:
raise HTTPException(status_code=404, detail="User not found")
# --- ML Model ---
class PredictionModel(MLResource):
name = "predictor"
verbose_name = "ML Predictor"
model_path = "sklearn_model.flm"
# --- MCP Server ---
mcp_server = MCPServer("tools", version="1.0.0", instructions="API tools for user management")
@mcp_server.tool("user_count", description="Get the total number of users")
async def user_count() -> int:
return 42
@mcp_server.resource("config://app", name="config", description="App config", mime_type="application/json")
def config():
return json.dumps({"version": "1.0.0"})
@mcp_server.prompt("welcome", description="Generate a welcome message")
def welcome(name: str):
return f"Write a friendly welcome message for {name}."
# --- App assembly ---
app = Flama(
openapi={
"info": {
"title": "Full-Stack ML + LLM + MCP + DDD API",
"version": "1.0.0",
"description": "Flama production API 🔥",
}
},
docs="/docs/",
components=[
WorkerComponent(worker=UserWorker()),
],
modules=[
SQLAlchemyModule(url="sqlite+aiosqlite:///app.db"),
],
)
app.resources.add_resource("/users/", UserResource)
app.models.add_model_resource(path="/predict", resource=PredictionModel)
app.models.add_model(path="/llm/", model="gemma.flm", name="gemma", serving=("native", "openai"))
app.mcp.add_server("/mcp/", "tools", server=mcp_server)
if __name__ == "__main__":
flama.run(flama_app="__main__:app", server_host="0.0.0.0", server_port=8080, server_reload=True)
