Skip to main content

​
Overview

The BaseApplication class is the central orchestration component of the Atlan Application SDK. It provides a standardized way to set up and run applications that use Temporal workflows, including automatic configuration of workflow clients, workers, and FastAPI servers.
The BaseApplication is designed to be used directly for most applications. For specialized use cases, you can subclass it to add custom functionality.

​
Key Features

  • Unified Workflow Management: Single interface for managing workflows, workers, and servers
  • Multi-Mode Deployment: Support for local development, worker-only, and server-only modes
  • Event-Based Workflows: Built-in support for event subscriptions and triggers
  • MCP Integration: Optional Model Context Protocol (MCP) server support
  • Application Manifest: Configuration-driven event registration

​
Initialization

from application_sdk.application import BaseApplication
from application_sdk.server import APIServer
from my_app.client import MyClient
from my_app.handler import MyHandler

app = BaseApplication(
    name="my-application",
    server=APIServer(),
    application_manifest=manifest_config,
    client_class=MyClient,
    handler_class=MyHandler
)

​
Constructor Parameters

​
name
str
required
The name of the application. Used for workflow client identification and logging.
​
server
ServerInterface
default:"None"
The server instance for the application. Typically an APIServer instance.
​
application_manifest
Dict[str, Any]
default:"None"
Application manifest configuration containing event registration and other settings.
application_manifest = {
    "eventRegistration": {
        "consumes": [
            {
                "event_type": "ENTITY_UPDATE",
                "event_name": "asset_updated",
                "event_id": "asset_update_trigger",
                "filters": []
            }
        ]
    }
}
​
client_class
Type[BaseClient]
default:"BaseClient"
Client class for the application. Must inherit from BaseClient.
​
handler_class
Type[BaseHandler]
default:"BaseHandler"
Handler class for the application. Must inherit from BaseHandler.

​
Core Methods

​
setup_workflow()

Set up the workflow client and initialize the worker with workflows and activities. 
await app.setup_workflow(
    workflow_and_activities_classes=[
        (MyWorkflow, MyActivities)
    ],
    passthrough_modules=["my_app", "pandas"],
    activity_executor=ThreadPoolExecutor(max_workers=10)
)
​
workflow_and_activities_classes
List[Tuple[Type[WorkflowInterface], Type[ActivitiesInterface]]]
required
List of tuples pairing workflow classes with their corresponding activities classes.
​
passthrough_modules
List[str]
default:"[]"
List of module names to pass through to the worker sandbox. These modules will be accessible within workflows.
​
activity_executor
ThreadPoolExecutor
default:"None"
Custom thread pool executor for running activities. If not provided, a default executor is created.

​
start()

Start the application based on the configured APPLICATION_MODE environment variable. 
await app.start(
    workflow_class=MyWorkflow,
    ui_enabled=True,
    has_configmap=False
)
​
workflow_class
Type[WorkflowInterface]
required
The main workflow class to register with the server.
​
ui_enabled
bool
default:"True"
Whether to enable the UI frontend routes.
​
has_configmap
bool
default:"False"
Whether the application has a configuration map for the UI.
The start() method automatically determines what to start based on APPLICATION_MODE:
  • LOCAL: Starts worker in daemon mode + server (for development)
  • WORKER: Starts only the worker (for production worker pods)
  • SERVER: Starts only the server (for production API server pods)

​
start_workflow()

Manually start a new workflow execution. 
result = await app.start_workflow(
    workflow_args={
        "workflow_id": "my-workflow-123",
        "metadata": {"key": "value"}
    },
    workflow_class=MyWorkflow
)
​
workflow_args
Dict[str, Any]
required
Arguments to pass to the workflow execution.
​
workflow_class
Type[WorkflowInterface]
required
The workflow class to execute.

​
Deployment Modes

The SDK supports three deployment modes via the APPLICATION_MODE environment variable:
Use case: Local developmentStarts both the worker (in daemon mode) and the server in a single process. The worker runs in the background while the server handles API requests.
export APPLICATION_MODE=LOCAL
python main.py
Use case: Production worker podsStarts only the worker in non-daemon mode. This is the mode to use for dedicated worker containers in production.
export APPLICATION_MODE=WORKER
python main.py
Use case: Production API server podsStarts only the FastAPI server without a worker. Use this for API-only containers that trigger workflows but don’t execute them.
export APPLICATION_MODE=SERVER
python main.py

​
Event Registration

Register event-based workflow triggers using the application manifest and registration API. 
# Define event subscription in manifest
manifest = {
    "eventRegistration": {
        "consumes": [
            {
                "event_type": "ENTITY_UPDATE",
                "event_name": "asset_updated",
                "event_id": "asset_update_handler",
                "filters": [
                    {
                        "path": "event.data.typeName",
                        "operator": "==",
                        "value": "Table"
                    }
                ]
            }
        ]
    }
}

app = BaseApplication(
    name="my-app",
    application_manifest=manifest
)

# Register workflow for the event
app.register_event_subscription(
    event_id="asset_update_handler",
    workflow_class=AssetUpdateWorkflow
)
Event IDs must be unique across the application manifest. Duplicate event IDs will raise a ValueError.

​
Complete Example

import asyncio
from application_sdk.application import BaseApplication
from my_app.workflows import MyWorkflow
from my_app.activities import MyActivities
from my_app.handlers import MyHandler
from my_app.clients import MyClient

async def main():
    # Initialize application
    app = BaseApplication(
        name="my-application",
        client_class=MyClient,
        handler_class=MyHandler
    )
    
    # Set up workflows and activities
    await app.setup_workflow(
        workflow_and_activities_classes=[
            (MyWorkflow, MyActivities)
        ],
        passthrough_modules=["my_app"]
    )
    
    # Start the application (mode determined by APPLICATION_MODE env var)
    await app.start(
        workflow_class=MyWorkflow,
        ui_enabled=True
    )

if __name__ == "__main__":
    asyncio.run(main())

​
Instance Attributes

AttributeTypeDescription
application_namestrName of the application
serverOptional[ServerInterface]Server instance for handling HTTP requests
workerOptional[Worker]Worker instance for executing workflows
workflow_clientWorkflowClientClient for interacting with Temporal
application_manifestOptional[Dict[str, Any]]Application configuration manifest
event_subscriptionsDict[str, EventWorkflowTrigger]Registered event subscriptions
client_classType[BaseClient]Client class for the application
handler_classType[BaseHandler]Handler class for the application
mcp_serverOptional[MCPServer]MCP server instance (if MCP is enabled)

Build docs developers (and LLMs) love

Get started for freeTalk to us