Skip to content

Services Package API Reference

ConnectionManager 

ConnectionManager(
    url: Url | None = None,
    server_id: UUID | None = None,
    server_pid_file: str | None = None,
)

Bases: Mindtrace

Client-side helper class for communicating with Mindtrace servers.

mcp_url property 
mcp_url: str

Return the MCP endpoint URL for this service instance using config paths.

mcp_client property 
mcp_client: Client

Get an MCP client for this service.

Returns a FastMCP Client instance that can be used to interact with the service through the MCP protocol. The client connects to the service's MCP endpoint.

Returns:

Type Description
Client

FastMCP Client instance for MCP protocol communication

Example:: cm = MyService.launch() client = cm.mcp_client # Use client for MCP protocol interactions

shutdown 
shutdown(block: bool = True)

Shutdown the server.

This method sends a shutdown request to the server. If block=True, it will also poll the server until it becomes unavailable, ensuring the shutdown process is complete.

Parameters:

Name Type Description Default
block bool

If True, waits for the server to actually shut down. If False, returns immediately after sending

 True

Example::

from mindtrace.services import Service, ServerStatus

cm = Service.launch()
assert cm.status == ServerStatus.Available

# Wait for shutdown to complete
cm.shutdown(block=True)
assert cm.status == ServerStatus.Down

# Or send shutdown command and return immediately
cm.shutdown(block=False)
ashutdown async 
ashutdown(block: bool = True)

Async shutdown of the server.

status 
status()

Get the status of the server.

Returns ServerStatus.DOWN if the server is unreachable, otherwise returns the actual status.

Returns:

Type Description

StatusOutput with the current server status.
astatus async 
astatus()

Async get the status of the server.

Returns ServerStatus.DOWN if the server is unreachable, otherwise returns the actual status.

Returns:

Type Description

StatusOutput with the current server status.

Service 

Service(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    summary: str | None = None,
    description: str | None = None,
    terms_of_service: str | None = None,
    license_info: Dict[str, str | Any] | None = None,
    live_service: bool = True,
    pid_file: str | None = None,
    **kwargs
)

Bases: Mindtrace

Base class for all Mindtrace services.

Initialize server instance. This is for internal use by the launch() method.

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object

 None
host str | None

Host address (e.g. "localhost" or "192.168.1.100")

 None
port int | None

Port number

 None
summary str | None

Summary of the server

 None
description str | None

Description of the server

 None
terms_of_service str | None

Terms of service for the server

 None
license_info Dict[str, str | Any] | None

License information for the server

 None
live_service bool

bool: set to True when launching via .launch(), set to False when querying endpoints in mindtrace.services.core.utils.py::generate_connection_manager Used to allow Service subclasses to have expensive init() methods without making .connect() slow

 True

Warning: Services should be created via the ServiceClass.launch() method. The init method here should be considered private internal use.

endpoints property 
endpoints: dict[str, TaskSchema]

Return the available commands for the service.

status property 
status: ServerStatus

Returns the current status of this service.

endpoints_func 
endpoints_func()

List all available endpoints for the service.

status_func 
status_func()

Get the current status of the service.

heartbeat_func 
heartbeat_func()

Perform a heartbeat check for the service.

status_at_host classmethod 
status_at_host(url: str | Url, timeout: int = 60) -> ServerStatus

Check the status of the service at the given host url.

This command may be used to check if a service (including this one) is available at a given host, useful for determining when a service has been successfully launched.

Parameters:

Name Type Description Default
url str | Url

The host URL of the service.

 required
connect classmethod 
connect(url: str | Url | None = None, timeout: int = 60) -> Any

Connect to an existing service.

The returned connection manager is determined by the registered connection manager for the service. If one has not explicitly been registered, the default connection manager (ConnectionManagerBase) will be used.

Parameters:

Name Type Description Default
url str | Url | None

The host URL of the service.

 None

Returns:

Type Description
Any

A connection manager for the service.

Raises:

Type Description
HTTPException

If the server fails to connect, an HTTPException will be raised with status code 503.
launch classmethod 
launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: Literal[False],
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
) -> None

launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: Literal[True] | bool = True,
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
) -> Any

launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: bool = True,
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
)

Launch a new server instance.

The server can be configured through either explicit URL parameters or through kwargs. All kwargs are passed directly to the server instance's init method.

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object (highest priority)

 None
host str | None

Host address (used if url not provided)

 None
port int | None

Port number (used if url not provided)

 None
block bool

If True, blocks the calling process and keeps the server running

 False
num_workers int

Number of worker processes

 1
wait_for_launch bool

Whether to wait for server startup

 True
timeout int

Timeout for server startup in seconds

 60
progress_bar bool

Show progress bar during startup

 True
**kwargs

Additional parameters passed to the server's init method

 {}
heartbeat 
heartbeat() -> Heartbeat

Request the server to do a complete heartbeat check.

shutdown staticmethod 
shutdown() -> fastapi.Response

HTTP endpoint to shut down the server.

shutdown_cleanup async 
shutdown_cleanup()

Cleanup the server.

Override this method in subclasses to shut down any additional resources (e.g. db connections) as necessary.

default_url classmethod 
default_url() -> Url

Get the default URL for this server type from config.

Priority:

  1. Server-specific URL from config
  2. Default ServerBase URL from config
  3. Fallback to localhost:8000
build_url classmethod 
build_url(
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
) -> Url

Build a URL with consistent priority logic.

Priority:

  1. Explicit URL parameter
  2. Host/port parameters
  3. Default URL from config

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object

 None
host str | None

Host address (e.g. "localhost" or "192.168.1.100")

 None
port int | None

Port number

 None

Returns:

Type Description
Url

Parsed URL object
get_mcp_paths classmethod 
get_mcp_paths() -> tuple[str, str]

Return (mount_path, http_app_path) for MCP based on config defaults.

Defaults: - mount_path: "/mcp-server" - http_app_path: "/mcp"

register_connection_manager classmethod 
register_connection_manager(connection_manager: Type[ConnectionManager])

Register a connection manager for this server.

default_log_file classmethod 
default_log_file() -> str

Get the default log file for this server type.

add_endpoint 
add_endpoint(
    path,
    func,
    schema: TaskSchema,
    api_route_kwargs=None,
    autolog_kwargs=None,
    methods: list[str] | None = None,
    scope: str = "public",
    as_tool: bool = False,
)

Register a new endpoint with optional role.

add_tool 
add_tool(tool_name, func)

Add a tool to the MCP server, with an informative description including the tool and service name.

Heartbeat dataclass 

Heartbeat(
    status: ServerStatus = ServerStatus.DOWN,
    server_id: UUID | None = None,
    message: str | None = None,
    details: Any = None,
)

Heartbeat status of a server.

Attributes:

Name Type Description
status ServerStatus

The current status of the server.
server_id UUID | None

The unique identifier of the server.
message str | None

Human-readable message describing the status of the server.
details Any

Additional details about the server status. Individual server subclasses may define their own specific protocol for this field (though always a dict). A GatewayServer, for instance, will return a dict[UUID, Heartbeat], containing the Heartbeats of all connected services, keyed by their unique server IDs.

Gateway 

Gateway(**kwargs)

Bases: Service

register_app 
register_app(payload: AppConfig)

Register a FastAPI app with the gateway.

forward_request async 
forward_request(request: Request, app_name: str, path: str)

Forward the request to the registered app.

connect classmethod 
connect(url: str | Url | None = None, timeout: int = 60) -> Any

Connect to an existing Gateway service with enhanced connection manager.

ProxyConnectionManager 

ProxyConnectionManager(
    gateway_url: str | Url, app_name: str, original_cm: ConnectionManager
)

A schema-aware proxy that forwards requests through the gateway instead of directly through the wrapped connection manager.

Initializes the ProxyConnectionManager.

Parameters:

Name Type Description Default
gateway_url str | Url

The base URL of the gateway.

 required
app_name str

The registered app name.

 required
original_cm ConnectionManager

The original connection manager.

 required

generate_connection_manager 

generate_connection_manager(
    service_cls,
    protected_methods: list[str] = [
        "shutdown",
        "ashutdown",
        "status",
        "astatus",
    ],
) -> type

Generates a dedicated ConnectionManager class with one method per endpoint.

Parameters:

Name Type Description Default
service_cls

The service class to generate a connection manager for.

 required
protected_methods list[str]

A list of methods that should not be overridden by dynamic methods.

 ['shutdown', 'ashutdown', 'status', 'astatus']

Returns:

Type Description
type

A ConnectionManager class with one method per endpoint.

core

connection_manager

Client-side helper class for communicating with any ServerBase server.

ConnectionManager 
ConnectionManager(
    url: Url | None = None,
    server_id: UUID | None = None,
    server_pid_file: str | None = None,
)

Bases: Mindtrace

Client-side helper class for communicating with Mindtrace servers.

mcp_url property 
mcp_url: str

Return the MCP endpoint URL for this service instance using config paths.

mcp_client property 
mcp_client: Client

Get an MCP client for this service.

Returns a FastMCP Client instance that can be used to interact with the service through the MCP protocol. The client connects to the service's MCP endpoint.

Returns:

Type Description
Client

FastMCP Client instance for MCP protocol communication

Example:: cm = MyService.launch() client = cm.mcp_client # Use client for MCP protocol interactions

shutdown 
shutdown(block: bool = True)

Shutdown the server.

This method sends a shutdown request to the server. If block=True, it will also poll the server until it becomes unavailable, ensuring the shutdown process is complete.

Parameters:

Name Type Description Default
block bool

If True, waits for the server to actually shut down. If False, returns immediately after sending

 True

Example::

from mindtrace.services import Service, ServerStatus

cm = Service.launch()
assert cm.status == ServerStatus.Available

# Wait for shutdown to complete
cm.shutdown(block=True)
assert cm.status == ServerStatus.Down

# Or send shutdown command and return immediately
cm.shutdown(block=False)
ashutdown async 
ashutdown(block: bool = True)

Async shutdown of the server.

status 
status()

Get the status of the server.

Returns ServerStatus.DOWN if the server is unreachable, otherwise returns the actual status.

Returns:

Type Description

StatusOutput with the current server status.
astatus async 
astatus()

Async get the status of the server.

Returns ServerStatus.DOWN if the server is unreachable, otherwise returns the actual status.

Returns:

Type Description

StatusOutput with the current server status.
launcher
Launcher 
Launcher(options)

Uvicorn application launcher for Mindtrace services (Windows).

mcp_client_manager
MCPClientManager 
MCPClientManager(service_cls: Type[Service])

Manager for MCP client operations for a service class.

Initialize the MCP client manager.

Parameters:

Name Type Description Default
service_cls Type[Service]

The service class this manager is bound to.

 required
connect 
connect(url: str | Url | None = None) -> Client

Connect to an existing service via MCP protocol.

Parameters:

Name Type Description Default
url str | Url | None

The URL of the service. If None, uses the default URL for this service type.

 None

Returns:

Type Description
Client

FastMCP Client instance for MCP protocol communication

Example:: from mindtrace.services.samples.echo_mcp import EchoService # Connect to a running EchoService mcp_client = EchoService.mcp.connect("http://localhost:8000")

# Use default URL
mcp_client = EchoService.mcp.connect()

# Use the connected client
async with mcp_client:
    tools = await mcp_client.list_tools()
    print(f"Available tools: {tools}")
launch 
launch(**launch_kwargs) -> Client

Launch a new service and return an MCP client for it.

Launches a new service instance using the service's launch method, then returns an MCP client connected to the newly launched service.

Parameters:

Name Type Description Default
**launch_kwargs

Arguments passed to the service's launch method

 {}

Returns:

Type Description
Client

FastMCP Client instance connected to the newly launched service

Example:: from mindtrace.services.samples.echo_mcp import EchoService

mcp_client = EchoService.mcp.launch(
                                    host="localhost",
                                    port=8000,
                                    wait_for_launch=True,
                                    timeout=10
                                    )

print(f"Service launched and MCP client created: {mcp_client}")
async with mcp_client:
    tools = await mcp_client.list_tools()
    print(f"Available tools: {tools}")
middleware
RequestLoggingMiddleware 
RequestLoggingMiddleware(
    app: Any,
    service_name: str,
    *,
    log_metrics: bool = False,
    metrics_interval: Optional[int] = None,
    metrics_to_collect: Optional[list[str]] = ["cpu_percent", "memory_percent"],
    add_request_id_header: bool = True,
    logger: Optional[Any] = None
)

Bases: BaseHTTPMiddleware

Minimal middleware for request-scoped logging without duplicating autolog.

Responsibilities: - Generate/bind a correlation id (request_id) via structlog.contextvars - Log one request-level envelope (request_started, request_completed) - Optionally log system metrics at request time - Attach request_id to response headers - Global error capture with structured logs

Avoids per-operation details already handled by @Mindtrace.autolog (duration_ms, per-endpoint metrics, start/completed of handlers).

service

Service base class. Provides unified methods for all Mindtrace (micro)services.

Service 
Service(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    summary: str | None = None,
    description: str | None = None,
    terms_of_service: str | None = None,
    license_info: Dict[str, str | Any] | None = None,
    live_service: bool = True,
    pid_file: str | None = None,
    **kwargs
)

Bases: Mindtrace

Base class for all Mindtrace services.

Initialize server instance. This is for internal use by the launch() method.

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object

 None
host str | None

Host address (e.g. "localhost" or "192.168.1.100")

 None
port int | None

Port number

 None
summary str | None

Summary of the server

 None
description str | None

Description of the server

 None
terms_of_service str | None

Terms of service for the server

 None
license_info Dict[str, str | Any] | None

License information for the server

 None
live_service bool

bool: set to True when launching via .launch(), set to False when querying endpoints in mindtrace.services.core.utils.py::generate_connection_manager Used to allow Service subclasses to have expensive init() methods without making .connect() slow

 True

Warning: Services should be created via the ServiceClass.launch() method. The init method here should be considered private internal use.

endpoints property 
endpoints: dict[str, TaskSchema]

Return the available commands for the service.

status property 
status: ServerStatus

Returns the current status of this service.

endpoints_func 
endpoints_func()

List all available endpoints for the service.

status_func 
status_func()

Get the current status of the service.

heartbeat_func 
heartbeat_func()

Perform a heartbeat check for the service.

status_at_host classmethod 
status_at_host(url: str | Url, timeout: int = 60) -> ServerStatus

Check the status of the service at the given host url.

This command may be used to check if a service (including this one) is available at a given host, useful for determining when a service has been successfully launched.

Parameters:

Name Type Description Default
url str | Url

The host URL of the service.

 required
connect classmethod 
connect(url: str | Url | None = None, timeout: int = 60) -> Any

Connect to an existing service.

The returned connection manager is determined by the registered connection manager for the service. If one has not explicitly been registered, the default connection manager (ConnectionManagerBase) will be used.

Parameters:

Name Type Description Default
url str | Url | None

The host URL of the service.

 None

Returns:

Type Description
Any

A connection manager for the service.

Raises:

Type Description
HTTPException

If the server fails to connect, an HTTPException will be raised with status code 503.
launch classmethod 
launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: Literal[False],
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
) -> None

launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: Literal[True] | bool = True,
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
) -> Any

launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: bool = True,
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
)

Launch a new server instance.

The server can be configured through either explicit URL parameters or through kwargs. All kwargs are passed directly to the server instance's init method.

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object (highest priority)

 None
host str | None

Host address (used if url not provided)

 None
port int | None

Port number (used if url not provided)

 None
block bool

If True, blocks the calling process and keeps the server running

 False
num_workers int

Number of worker processes

 1
wait_for_launch bool

Whether to wait for server startup

 True
timeout int

Timeout for server startup in seconds

 60
progress_bar bool

Show progress bar during startup

 True
**kwargs

Additional parameters passed to the server's init method

 {}
heartbeat 
heartbeat() -> Heartbeat

Request the server to do a complete heartbeat check.

shutdown staticmethod 
shutdown() -> fastapi.Response

HTTP endpoint to shut down the server.

shutdown_cleanup async 
shutdown_cleanup()

Cleanup the server.

Override this method in subclasses to shut down any additional resources (e.g. db connections) as necessary.

default_url classmethod 
default_url() -> Url

Get the default URL for this server type from config.

Priority:

  1. Server-specific URL from config
  2. Default ServerBase URL from config
  3. Fallback to localhost:8000
build_url classmethod 
build_url(
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
) -> Url

Build a URL with consistent priority logic.

Priority:

  1. Explicit URL parameter
  2. Host/port parameters
  3. Default URL from config

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object

 None
host str | None

Host address (e.g. "localhost" or "192.168.1.100")

 None
port int | None

Port number

 None

Returns:

Type Description
Url

Parsed URL object
get_mcp_paths classmethod 
get_mcp_paths() -> tuple[str, str]

Return (mount_path, http_app_path) for MCP based on config defaults.

Defaults: - mount_path: "/mcp-server" - http_app_path: "/mcp"

register_connection_manager classmethod 
register_connection_manager(connection_manager: Type[ConnectionManager])

Register a connection manager for this server.

default_log_file classmethod 
default_log_file() -> str

Get the default log file for this server type.

add_endpoint 
add_endpoint(
    path,
    func,
    schema: TaskSchema,
    api_route_kwargs=None,
    autolog_kwargs=None,
    methods: list[str] | None = None,
    scope: str = "public",
    as_tool: bool = False,
)

Register a new endpoint with optional role.

add_tool 
add_tool(tool_name, func)

Add a tool to the MCP server, with an informative description including the tool and service name.

types
Heartbeat dataclass 
Heartbeat(
    status: ServerStatus = ServerStatus.DOWN,
    server_id: UUID | None = None,
    message: str | None = None,
    details: Any = None,
)

Heartbeat status of a server.

Attributes:

Name Type Description
status ServerStatus

The current status of the server.
server_id UUID | None

The unique identifier of the server.
message str | None

Human-readable message describing the status of the server.
details Any

Additional details about the server status. Individual server subclasses may define their own specific protocol for this field (though always a dict). A GatewayServer, for instance, will return a dict[UUID, Heartbeat], containing the Heartbeats of all connected services, keyed by their unique server IDs.
utils
add_endpoint 
add_endpoint(app, path, self: Optional[Service], **kwargs)

Register a new endpoint.

This decorator method is functionally identical as calling add_endpoint on a Service instance. It is useful when the endpoints are defined in a separate method, such as grouping api routes in a more complicated FastAPI app.

Parameters:

Name Type Description Default
app

The FastAPI app.

 required
path

The endpoint path.

 required
self Optional[Service]

The server instance.

 required
**kwargs

Additional arguments to pass when creating the FastAPI route.

 {}

Example::

from fastapi import FastAPI
from mindtrace.services import Service

class MyServer(Service):
    def __init__(self):
        super().__init__()

        self.add_endpoint(path="/status_using_method", func=self.status)
        self.create_app()

    def status(self):
        return {"status": "Available"}

    def create_app():
        # May put all the endpoints in a single method, and call the method in __init__.

        @add_endpoint(self.app, "/status_using_decorator", self=self)
        def status():
            return {"status": "Available"}

        @add_endpoint(self.app, "/another_hundred_endpoints", self=self)
        def another_hundred_endpoints():
            return
register_connection_manager 
register_connection_manager(connection_manager: Type[ConnectionManager])

Register a connection manager for a server class.

This decorator is used to register a connection manager for a server class. The connection manager is used to communicate with the server. The connection manager must be a subclass of ConnectionManager.

Parameters:

Name Type Description Default
connection_manager Type[ConnectionManager]

The connection manager class.

 required

Example::

import requests
from mindtrace.services import ConnectionManager, Service

class MyConnectionManager(ConnectionManager):
    def __init__(self, url):
        super().__init__(url)

    def add(arg1, arg2):
        response = requests.request("POST", str(self.url) + "add", json={"arg1": arg1, "arg2": arg2})
        return json.loads(response.content)["sum"]

@register_connection_manager(MyConnectionManager)
class MyService(Service):
    def __init__(self):
        super().__init__()
        self.add_endpoint("add", self.add)

    def add(self, arg1, arg2):
        return {"sum": arg1 + arg2}

cm = MyService.launch()  # Returns a MyConnectionManager instance, NOT a MyServer instance
sum = cm.add(1, 2)  # Calls add method in MyConnectionManager
generate_connection_manager 
generate_connection_manager(
    service_cls,
    protected_methods: list[str] = [
        "shutdown",
        "ashutdown",
        "status",
        "astatus",
    ],
) -> type

Generates a dedicated ConnectionManager class with one method per endpoint.

Parameters:

Name Type Description Default
service_cls

The service class to generate a connection manager for.

 required
protected_methods list[str]

A list of methods that should not be overridden by dynamic methods.

 ['shutdown', 'ashutdown', 'status', 'astatus']

Returns:

Type Description
type

A ConnectionManager class with one method per endpoint.

discord

Discord integration for Mindtrace services.

DiscordClient 
DiscordClient(
    *, token: str | None = None, intents: Optional[Intents] = None, **kwargs
)

Bases: Mindtrace

Discord client that can be extended for different bot implementations.

This class provides: - Command registration and management - Event handling system - Integration with Mindtrace patterns - Configurable bot behavior

Initialize the Discord client.

Parameters:

Name Type Description Default
token str | None

Discord bot token (optional, will use config if not provided)

 None
intents Optional[Intents]

Discord intents configuration

 None
**kwargs

Additional arguments passed to Mindtrace

 {}
register_command 
register_command(
    name: str,
    description: str,
    usage: str,
    handler: Callable,
    aliases: Optional[List[str]] = None,
    category: str = "General",
    enabled: bool = True,
    hidden: bool = False,
    cooldown: Optional[int] = None,
    permissions: Optional[List[str]] = None,
    parameters: Optional[Dict[str, Dict[str, Any]]] = None,
)

Register a new slash command with the bot.

Parameters:

Name Type Description Default
name str

Command name

 required
description str

Command description

 required
usage str

Usage instructions (for documentation)

 required
handler Callable

Function to handle the command

 required
aliases Optional[List[str]]

Alternative command names (not used in slash commands)

 None
category str

Command category

 'General'
enabled bool

Whether the command is enabled

 True
hidden bool

Whether to hide from help

 False
cooldown Optional[int]

Cooldown in seconds

 None
permissions Optional[List[str]]

Required permissions

 None
parameters Optional[Dict[str, Dict[str, Any]]]

Dictionary of parameter descriptions for slash commands

 None
register_event_handler 
register_event_handler(
    event_type: DiscordEventType, handler: DiscordEventHandler
)

Register an event handler.

Parameters:

Name Type Description Default
event_type DiscordEventType

Type of event to handle

 required
handler DiscordEventHandler

Event handler instance

 required
start_bot async 
start_bot()

Start the Discord bot.

stop_bot async 
stop_bot()

Stop the Discord bot.

DiscordService 
DiscordService(
    *, token: str | None = None, intents: Optional[Any] = None, **kwargs
)

Bases: Service

Service wrapper for DiscordClient.

This class provides: - HTTP API endpoints for Discord bot control - MCP tool integration - Service lifecycle management - Integration with Mindtrace infrastructure

Initialize the Discord service.

Parameters:

Name Type Description Default
token str | None

Discord bot token (optional, will use config if not provided)

 None
intents Optional[Any]

Discord intents configuration

 None
**kwargs

Additional arguments passed to Service

 {}
startup async 
startup()

Startup the Discord bot during service initialization.

execute_command async 
execute_command(payload: DiscordCommandInput) -> DiscordCommandOutput

Execute a command via the service API.

This method allows executing Discord slash commands programmatically through the FastAPI endpoint, useful for exposing AI models and other functionality through both Discord and HTTP interfaces.

Parameters:

Name Type Description Default
payload DiscordCommandInput

Command input data

 required

Returns:

Type Description
DiscordCommandOutput

Command output
get_bot_status 
get_bot_status() -> DiscordStatusOutput

Get the current bot status.

Returns:

Type Description
DiscordStatusOutput

Bot status information
get_commands 
get_commands() -> DiscordCommandsOutput

Get list of registered commands.

Returns:

Type Description
DiscordCommandsOutput

Command information
register_command 
register_command(*args, **kwargs)

Register a command with the Discord client.

register_event_handler 
register_event_handler(*args, **kwargs)

Register an event handler with the Discord client.

launch classmethod 
launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: bool = True,
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
)

Launch a Discord service and wait for the Discord bot to be ready.

This overrides the base Service.launch() method to ensure the Discord bot is fully connected before returning the connection manager.

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object (highest priority)

 None
host str | None

Host address (used if url not provided)

 None
port int | None

Port number (used if url not provided)

 None
block bool

If True, blocks the calling process and keeps the server running

 False
num_workers int

Number of worker processes

 1
wait_for_launch bool

Whether to wait for server startup

 True
timeout int

Timeout for server startup in seconds

 60
progress_bar bool

Show progress bar during startup

 True
**kwargs

Additional parameters passed to the server's init method

 {}
shutdown_cleanup async 
shutdown_cleanup()

Cleanup when shutting down the service.

DiscordCommand dataclass 
DiscordCommand(
    name: str,
    description: str,
    usage: str,
    aliases: List[str],
    category: str,
    enabled: bool = True,
    hidden: bool = False,
    cooldown: Optional[int] = None,
    permissions: Optional[List[str]] = None,
)

Represents a Discord command with its metadata.

DiscordCommandInput

Bases: BaseModel

Base input schema for Discord commands.

DiscordCommandOutput

Bases: BaseModel

Base output schema for Discord commands.

DiscordCommandSchema

Bases: TaskSchema

Base schema for Discord commands.

DiscordCommandsOutput

Bases: BaseModel

Output schema for commands list.

DiscordCommandsSchema

Bases: TaskSchema

Schema for commands list endpoint.

DiscordEventHandler

Bases: ABC

Abstract base class for Discord event handlers.

handle abstractmethod async 
handle(event_type: DiscordEventType, **kwargs) -> None

Handle a Discord event.

DiscordEventType

Bases: Enum

Types of Discord events that can be handled.

DiscordStatusOutput

Bases: BaseModel

Output schema for bot status.

DiscordStatusSchema

Bases: TaskSchema

Schema for bot status endpoint.

discord_client

Discord client implementation for Mindtrace services.

This module provides a base Discord client that can be extended for different bot implementations. It follows the Mindtrace Service patterns and provides a clean interface for command registration.

DiscordClient 
DiscordClient(
    *, token: str | None = None, intents: Optional[Intents] = None, **kwargs
)

Bases: Mindtrace

Discord client that can be extended for different bot implementations.

This class provides: - Command registration and management - Event handling system - Integration with Mindtrace patterns - Configurable bot behavior

Initialize the Discord client.

Parameters:

Name Type Description Default
token str | None

Discord bot token (optional, will use config if not provided)

 None
intents Optional[Intents]

Discord intents configuration

 None
**kwargs

Additional arguments passed to Mindtrace

 {}
register_command 
register_command(
    name: str,
    description: str,
    usage: str,
    handler: Callable,
    aliases: Optional[List[str]] = None,
    category: str = "General",
    enabled: bool = True,
    hidden: bool = False,
    cooldown: Optional[int] = None,
    permissions: Optional[List[str]] = None,
    parameters: Optional[Dict[str, Dict[str, Any]]] = None,
)

Register a new slash command with the bot.

Parameters:

Name Type Description Default
name str

Command name

 required
description str

Command description

 required
usage str

Usage instructions (for documentation)

 required
handler Callable

Function to handle the command

 required
aliases Optional[List[str]]

Alternative command names (not used in slash commands)

 None
category str

Command category

 'General'
enabled bool

Whether the command is enabled

 True
hidden bool

Whether to hide from help

 False
cooldown Optional[int]

Cooldown in seconds

 None
permissions Optional[List[str]]

Required permissions

 None
parameters Optional[Dict[str, Dict[str, Any]]]

Dictionary of parameter descriptions for slash commands

 None
register_event_handler 
register_event_handler(
    event_type: DiscordEventType, handler: DiscordEventHandler
)

Register an event handler.

Parameters:

Name Type Description Default
event_type DiscordEventType

Type of event to handle

 required
handler DiscordEventHandler

Event handler instance

 required
start_bot async 
start_bot()

Start the Discord bot.

stop_bot async 
stop_bot()

Stop the Discord bot.

discord_service

Discord Service implementation for Mindtrace services.

This module provides a Service wrapper around DiscordClient that enables HTTP API endpoints and MCP integration while maintaining the Discord bot functionality.

DiscordService 
DiscordService(
    *, token: str | None = None, intents: Optional[Any] = None, **kwargs
)

Bases: Service

Service wrapper for DiscordClient.

This class provides: - HTTP API endpoints for Discord bot control - MCP tool integration - Service lifecycle management - Integration with Mindtrace infrastructure

Initialize the Discord service.

Parameters:

Name Type Description Default
token str | None

Discord bot token (optional, will use config if not provided)

 None
intents Optional[Any]

Discord intents configuration

 None
**kwargs

Additional arguments passed to Service

 {}
startup async 
startup()

Startup the Discord bot during service initialization.

execute_command async 
execute_command(payload: DiscordCommandInput) -> DiscordCommandOutput

Execute a command via the service API.

This method allows executing Discord slash commands programmatically through the FastAPI endpoint, useful for exposing AI models and other functionality through both Discord and HTTP interfaces.

Parameters:

Name Type Description Default
payload DiscordCommandInput

Command input data

 required

Returns:

Type Description
DiscordCommandOutput

Command output
get_bot_status 
get_bot_status() -> DiscordStatusOutput

Get the current bot status.

Returns:

Type Description
DiscordStatusOutput

Bot status information
get_commands 
get_commands() -> DiscordCommandsOutput

Get list of registered commands.

Returns:

Type Description
DiscordCommandsOutput

Command information
register_command 
register_command(*args, **kwargs)

Register a command with the Discord client.

register_event_handler 
register_event_handler(*args, **kwargs)

Register an event handler with the Discord client.

launch classmethod 
launch(
    *,
    url: str | Url | None = None,
    host: str | None = None,
    port: int | None = None,
    block: bool = False,
    num_workers: int = 1,
    wait_for_launch: bool = True,
    timeout: int = 60,
    progress_bar: bool = True,
    **kwargs
)

Launch a Discord service and wait for the Discord bot to be ready.

This overrides the base Service.launch() method to ensure the Discord bot is fully connected before returning the connection manager.

Parameters:

Name Type Description Default
url str | Url | None

Full URL string or Url object (highest priority)

 None
host str | None

Host address (used if url not provided)

 None
port int | None

Port number (used if url not provided)

 None
block bool

If True, blocks the calling process and keeps the server running

 False
num_workers int

Number of worker processes

 1
wait_for_launch bool

Whether to wait for server startup

 True
timeout int

Timeout for server startup in seconds

 60
progress_bar bool

Show progress bar during startup

 True
**kwargs

Additional parameters passed to the server's init method

 {}
shutdown_cleanup async 
shutdown_cleanup()

Cleanup when shutting down the service.

types
DiscordEventType

Bases: Enum

Types of Discord events that can be handled.

DiscordCommand dataclass 
DiscordCommand(
    name: str,
    description: str,
    usage: str,
    aliases: List[str],
    category: str,
    enabled: bool = True,
    hidden: bool = False,
    cooldown: Optional[int] = None,
    permissions: Optional[List[str]] = None,
)

Represents a Discord command with its metadata.

DiscordCommandInput

Bases: BaseModel

Base input schema for Discord commands.

DiscordCommandOutput

Bases: BaseModel

Base output schema for Discord commands.

DiscordStatusOutput

Bases: BaseModel

Output schema for bot status.

DiscordCommandsOutput

Bases: BaseModel

Output schema for commands list.

DiscordCommandSchema

Bases: TaskSchema

Base schema for Discord commands.

DiscordStatusSchema

Bases: TaskSchema

Schema for bot status endpoint.

DiscordCommandsSchema

Bases: TaskSchema

Schema for commands list endpoint.

DiscordEventHandler

Bases: ABC

Abstract base class for Discord event handlers.

handle abstractmethod async 
handle(event_type: DiscordEventType, **kwargs) -> None

Handle a Discord event.

gateway

gateway
Gateway 
Gateway(**kwargs)

Bases: Service

register_app 
register_app(payload: AppConfig)

Register a FastAPI app with the gateway.

forward_request async 
forward_request(request: Request, app_name: str, path: str)

Forward the request to the registered app.

connect classmethod 
connect(url: str | Url | None = None, timeout: int = 60) -> Any

Connect to an existing Gateway service with enhanced connection manager.

proxy_connection_manager
ProxyConnectionManager 
ProxyConnectionManager(
    gateway_url: str | Url, app_name: str, original_cm: ConnectionManager
)

A schema-aware proxy that forwards requests through the gateway instead of directly through the wrapped connection manager.

Initializes the ProxyConnectionManager.

Parameters:

Name Type Description Default
gateway_url str | Url

The base URL of the gateway.

 required
app_name str

The registered app name.

 required
original_cm ConnectionManager

The original connection manager.

 required