A reusable, abstract controller layer that standardizes the interface between backend logic and any UI surface — Streamlit, FastAPI, CLI, or any custom frontend.
Building applications that need to run across multiple interfaces (a web dashboard, a REST API, and a CLI) typically means duplicating the same business logic with framework-specific wiring at every layer. Each interface re-implements session management, state handling, caching, error normalization, and event routing.
ui-controller-base solves this by introducing a single controller that owns
all business logic. Thin adapters translate each UI framework's native
primitives into framework-agnostic value objects. The same controller class powers
every surface with zero duplication.
┌────────────────────────────────────────────────────────┐
│ AbstractUIController │
│ views · tools · state · cache · events │
└────────────┬─────────────┬──────────────┬──────────────┘
│ │ │
StreamlitAdapter FastAPIAdapter CLIAdapter
This package integrates 13 sibling packages from the same project family.
Each one handles one infrastructure concern; ui-controller-base orchestrates them:
| Package | What it provides | Used for |
|---|---|---|
abstract-agent |
AbstractAgent lifecycle, mixin patterns |
AbstractUIController inherits its lifecycle & mixin conventions |
plugin-loader |
AbstractPlugin, PluginRegistry, discovery strategies |
AbstractView is a plugin; ViewRegistry manages view lifecycle |
tool-interface-adapter |
AbstractTool, ToolRegistry, retry/cache/logging middleware |
Every backend operation is a UITool; registry routes actions |
event-dispatcher |
LocalDispatcher, BufferedAsyncDispatcher, event bus |
UIDispatcher emits ui.action.* events; ObservableMixin / ReactiveMixin subscribe |
logger-pkg |
Structured logging with request/session context binding | Per-action request_id and session_id threading |
error-exception-handler |
BaseAppException, ErrorHandler |
Exceptions are normalized to typed UIResponse.error(...) — never raw stack traces |
cache-manager |
LayeredCache, CacheNamespace, stampede protection |
CacheableMixin exposes ui.session, ui.view, ui.tool cache namespaces |
job-manager |
AsyncScheduler, MemoryJobStore, ImmediateTrigger |
WithJobs mixin — submit background jobs, get UIResponse.pending(job_id) |
workflow-manager |
WorkflowEngine, HumanStep, FunctionStep |
WithWorkflows mixin — multi-step flows, human-in-the-loop approval |
notification-manager |
NotificationManager, ChannelPlugin |
WithNotifications mixin — push status/alerts from controllers |
api-client-base |
AsyncAPIClient, auth schemes, retry/circuit-breaker |
WithApiClients mixin — named HTTP clients shared across views/tools |
db-accessor |
get_database, AbstractDatabase |
WithDatabase mixin — controller-level database access |
config-manager |
Config file management | ui-controller init scaffolder — generates consumer project config |
# From this directory with all adapters and capability mixins
pip install -e ".[all]"
# Minimal install (core only, no framework adapters)
pip install -e "."
# Selective installs
pip install -e ".[fastapi,cli]" # FastAPI + CLI adapters
pip install -e ".[streamlit]" # Streamlit adapter
pip install -e ".[jobs,workflows]" # Async jobs + multi-step workflowsA view renders a UI fragment for a given route. It is a plugin in the
plugin-loader sense — it has a lifecycle and is discovered automatically.
# myapp/views/home_view.py
from ui_controller_base import UIResponse
from ui_controller_base.core.context import UIContext
from ui_controller_base.core.response import UIFragment
from ui_controller_base.core.state import UIState
from ui_controller_base.core.view import AbstractView
from ui_controller_base.types import FragmentType
class HomeView(AbstractView):
plugin_name = "home"
route = "/home"
plugin_description = "Welcome page"
async def render(self, ctx: UIContext, state: UIState) -> UIResponse:
return UIResponse.ok(
UIFragment(type=FragmentType.TEXT, content="## Welcome"),
UIFragment(
type=FragmentType.FORM,
content={
"action": "greet",
"fields": [{"name": "name", "type": "text", "label": "Your name"}],
"submit_label": "Greet me!",
},
),
)# myapp/controller.py
from typing import Any
from ui_controller_base import (
AbstractUIController,
CacheableMixin,
ObservableMixin,
StatefulMixin,
UIResponse,
UITool,
)
from myapp.views.home_view import HomeView
async def greet(name: str = "World", **_: Any) -> UIResponse:
"""Return a personalised greeting."""
return UIResponse.text(f"Hello, {name}!", toast=f"Greeted {name}")
class MyController(CacheableMixin, StatefulMixin, ObservableMixin, AbstractUIController):
views = [HomeView]
tools = [UITool.from_fn("greet", greet, description="Greet a user by name")]Streamlit — runs as a standard Streamlit app; reruns are neutralized via
cache-manager so the view is rendered once and cached per session:
# myapp/run_streamlit.py
from ui_controller_base import StreamlitAdapter
from myapp.controller import MyController
StreamlitAdapter(MyController(), title="My App").serve()streamlit run myapp/run_streamlit.pyFastAPI — mounts views as GET routes and tools as POST action routes; full OpenAPI schema generated from pydantic value objects:
# myapp/run_fastapi.py
from fastapi import FastAPI
from ui_controller_base import FastAPIAdapter
from myapp.controller import MyController
app = FastAPI(title="My App")
FastAPIAdapter(MyController()).mount_app(app)uvicorn myapp.run_fastapi:app
# GET /home → render HomeView
# POST /home/action/greet → dispatch greet tool
# GET /health → health checkCLI — each registered tool becomes a typed Typer command; option flags are derived from the tool function's type-annotated signature:
# myapp/run_cli.py
from ui_controller_base import CLIAdapter
from myapp.controller import MyController
CLIAdapter(MyController()).serve()python myapp/run_cli.py greet --name Ada
python myapp/run_cli.py --help# Generate controller + views + entry points for all three adapters
ui-controller init MyApp --out ./projects
# List registered views and tools on an existing controller
ui-controller list-views myapp.controller.MyController
# Run adapters via the CLI helper
ui-controller run-streamlit myapp/run_streamlit.py
ui-controller run-fastapi myapp.run_fastapi:appCompose additional infrastructure onto AbstractUIController by listing
mixins left-to-right in the class definition (outermost first):
class MyController(
WithJobs, # background job scheduling (job-manager)
WithWorkflows, # multi-step flows with human approval (workflow-manager)
WithNotifications, # push alerts (notification-manager)
WithApiClients, # named HTTP clients (api-client-base)
WithDatabase, # database access (db-accessor)
ObservableMixin, # event bus (event-dispatcher)
StatefulMixin, # per-session UIState (optimistic locking)
CacheableMixin, # view/tool/session caches (cache-manager)
AbstractUIController,
):
...| Mixin | Adds | Backed by |
|---|---|---|
StatefulMixin |
UIState per session, InMemoryStateStore / FileStateStore |
built-in |
CacheableMixin |
ui.session / ui.view / ui.tool namespaced caches |
cache-manager |
ObservableMixin |
self.events event bus, ui.action.* events |
event-dispatcher |
ReactiveMixin |
Subscribe to backend.* events for live updates |
event-dispatcher |
WithJobs |
submit_job(job) → UIResponse.pending(job_id) |
job-manager |
WithWorkflows |
start_workflow / resume_workflow with HumanStep |
workflow-manager |
WithNotifications |
send_notification(title, body, recipient_id) |
notification-manager |
WithApiClients |
register_api_client(name, config) / get_api_client(name) |
api-client-base |
WithDatabase |
self.db — AbstractDatabase instance |
db-accessor |
| Object | Purpose |
|---|---|
UIAction(name, payload, session_id) |
Inbound action from a UI surface |
UIResponse(status, fragments, toast, errors) |
Outbound render instruction |
UIFragment(type, content) |
Single renderable unit (TEXT, TABLE, CHART, FORM, PROGRESS, NOTIFICATION, ERROR, HTML) |
UIState(session_id, data, version) |
Per-session mutable state |
UIContext(request_id, session_id, events, ...) |
Per-request read-only context |
Use InMemoryAdapter or ControllerTestHarness for fast, framework-free tests:
from ui_controller_base.testing import ControllerTestHarness
async def test_greet():
async with ControllerTestHarness(MyController) as adapter:
resp = await adapter.dispatch("greet", name="Ada")
adapter.assert_ok(resp)
assert "Ada" in resp.fragments[0].content
async def test_view_renders():
async with ControllerTestHarness(MyController) as adapter:
resp = await adapter.render_view("/home")
adapter.assert_ok(resp)pytest -q # run all tests
pytest --cov=ui_controller_base # with coverage report (threshold: 80%)A fully-working example lives in examples/minimal_controller/:
# CLI
python examples/minimal_controller/run_cli.py greet --name Ada
# FastAPI
uvicorn examples.minimal_controller.run_fastapi:app
curl -X POST localhost:8000/home/action/greet -H "Content-Type: application/json" \
-d '{"payload": {"name": "Ada"}}'
# Streamlit
streamlit run examples/minimal_controller/run_streamlit.py