1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
"""Starlette application factory for ollama-web."""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import Response
from starlette.routing import Mount, Route
from starlette.staticfiles import StaticFiles
from .auth import AuthMiddleware
from .config import settings
from .middleware import SecurityHeadersMiddleware
from .routes import auth as auth_route
from .routes import chat as chat_route
from .routes import mcp as mcp_route
from .routes import models as models_route
from .routes import pages as pages_route
from .routes import sessions as sessions_route
from .routes import settings_route
from .sessions import SessionStore
class NoCacheStaticFiles(StaticFiles):
"""StaticFiles variant that disables browser caching for local development.
This ensures language changes and JS updates are reflected immediately after
a server restart, without requiring users to clear the browser cache.
"""
async def get_response(self, path: str, scope: Any) -> Response:
response = await super().get_response(path, scope)
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
return response
_STATIC_DIR = Path(__file__).resolve().parent / "static"
@asynccontextmanager
async def _lifespan(app: Starlette) -> AsyncIterator[None]:
"""Keep the application log handler scoped to the ASGI lifespan."""
log_dir = Path(app.state.settings.data_dir).resolve()
log_dir.mkdir(parents=True, exist_ok=True)
handler = logging.FileHandler(str(log_dir / "server.log"), encoding="utf-8")
handler.setFormatter(
logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
)
logger = logging.getLogger("ollama_web")
previous_level = logger.level
logger.setLevel(logging.INFO)
logger.addHandler(handler)
try:
yield
finally:
logger.removeHandler(handler)
handler.close()
logger.setLevel(previous_level)
def create_app() -> Starlette:
"""Build and return the configured Starlette application."""
routes = [
Route("/login", pages_route.login, name="login_page"),
Route("/api/auth/login", auth_route.login, methods=["POST"], name="auth_login"),
Route("/api/auth/logout", auth_route.logout, methods=["POST"], name="auth_logout"),
Route("/api/auth/status", auth_route.status, methods=["GET"], name="auth_status"),
Route("/", pages_route.index, name="index"),
Route("/api/chat", chat_route.chat, methods=["POST"], name="chat"),
Route("/api/models", models_route.get_models, methods=["GET"], name="models"),
Route(
"/api/models/{model:path}/capabilities",
models_route.get_model_capabilities,
methods=["GET"],
name="model_capabilities",
),
Route("/api/health", models_route.health, methods=["GET"], name="health"),
Route(
"/api/settings",
settings_route.get_settings,
methods=["GET"],
name="settings",
),
Route(
"/api/settings",
settings_route.put_settings,
methods=["PUT"],
name="settings_update",
),
Route("/api/sessions", sessions_route.list_sessions, methods=["GET"], name="list_sessions"),
Route(
"/api/sessions",
sessions_route.create_session,
methods=["POST"],
name="create_session",
),
Route(
"/api/sessions/{session_id}",
sessions_route.get_session,
methods=["GET"],
name="get_session",
),
Route(
"/api/sessions/{session_id}",
sessions_route.delete_session,
methods=["DELETE"],
name="delete_session",
),
Route(
"/api/sessions/{session_id}/files",
sessions_route.upload_file,
methods=["POST"],
name="upload_file",
),
Route(
"/api/sessions/{session_id}/files/{file_id}",
sessions_route.delete_file,
methods=["DELETE"],
name="delete_file",
),
Route(
"/api/sessions/{session_id}/files/{file_id}/content",
sessions_route.get_file_content,
methods=["GET", "HEAD"],
name="get_file_content",
),
Route(
"/api/sessions/{session_id}/messages",
sessions_route.clear_messages,
methods=["DELETE"],
name="clear_messages",
),
Route("/api/mcp/servers", mcp_route.get_servers, methods=["GET"], name="mcp_servers"),
Route(
"/api/mcp/servers",
mcp_route.put_servers,
methods=["PUT"],
name="mcp_servers_update",
),
Mount("/static", app=NoCacheStaticFiles(directory=str(_STATIC_DIR)), name="static"),
]
middleware = [Middleware(SecurityHeadersMiddleware), Middleware(AuthMiddleware)]
if settings.allowed_origins:
middleware.append(
Middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Content-Type", "X-CSRF-Token"],
allow_credentials=True,
)
)
app = Starlette(routes=routes, middleware=middleware, lifespan=_lifespan)
app.state.settings = settings
app.state.session_store = SessionStore(settings.data_dir)
app.state.tool_names = ["web_search", "scrape_url", "search_and_fetch", "pdf_to_text"]
return app
app = create_app()
|
