Skip to content

Channels

octomate.config.channels

Channel tentacle configuration: one variant per platform, the streaming and chat-recap settings they share, and the OAuth clients a channel authorizes users with.

AgentModelConfig pydantic-model

Bases: BaseModel

An agent id paired with a model selection.

Fields:

Source code in octomate/config/channels.py
class AgentModelConfig(BaseModel):
    """An agent id paired with a model selection."""

    agent: str
    model: AgentRouteModelName | None = Field(
        description="Selected model, or null to use the harness's native default."
    )

agent pydantic-field

agent: str

model pydantic-field

model: AgentRouteModelName | None

Selected model, or null to use the harness's native default.

ChannelStreamConfig pydantic-model

Bases: BaseModel

How a channel paces streamed text: whether it streams, the flush cadence, the size bounds, and when a block folds.

Fields:

Source code in octomate/config/channels.py
class ChannelStreamConfig(BaseModel):
    """How a channel paces streamed text: whether it streams, the flush cadence,
    the size bounds, and when a block folds."""

    enabled: bool = False
    flush_interval: float = 0.5
    min_chars: int = 20
    max_chars: int = 1000
    fold_threshold: int = 1500

enabled pydantic-field

enabled: bool = False

flush_interval pydantic-field

flush_interval: float = 0.5

min_chars pydantic-field

min_chars: int = 20

max_chars pydantic-field

max_chars: int = 1000

fold_threshold pydantic-field

fold_threshold: int = 1500

SlackStreamConfig pydantic-model

Bases: ChannelStreamConfig

Streaming defaults for Slack.

Fields:

Source code in octomate/config/channels.py
class SlackStreamConfig(ChannelStreamConfig):
    """Streaming defaults for Slack."""

    # Slack streams via `chat.appendStream`, one API call per flush. `flush_interval`
    # alone paces the edits (~2/s) — keep `min_chars` low so short answers stream on
    # that cadence instead of being held until they reach 120 chars.
    enabled: bool = True
    flush_interval: float = 0.2
    min_chars: int = 20

enabled pydantic-field

enabled: bool = True

flush_interval pydantic-field

flush_interval: float = 0.2

min_chars pydantic-field

min_chars: int = 20

LarkStreamConfig pydantic-model

Bases: ChannelStreamConfig

Streaming defaults for Lark.

Fields:

Source code in octomate/config/channels.py
class LarkStreamConfig(ChannelStreamConfig):
    """Streaming defaults for Lark."""

    enabled: bool = True
    flush_interval: float = 0.2
    min_chars: int = 20

enabled pydantic-field

enabled: bool = True

flush_interval pydantic-field

flush_interval: float = 0.2

min_chars pydantic-field

min_chars: int = 20

DiscordStreamConfig pydantic-model

Bases: ChannelStreamConfig

Streaming defaults for Discord.

Fields:

Source code in octomate/config/channels.py
class DiscordStreamConfig(ChannelStreamConfig):
    """Streaming defaults for Discord."""

    # discord.py follows Discord's dynamic route buckets; this only coalesces edits.
    flush_interval: float = 0.2

flush_interval pydantic-field

flush_interval: float = 0.2

ChatRecapConfig pydantic-model

Bases: BaseModel

What a kick in a dm or a group chat is shown of the chat it woke in.

Those surfaces have no end, so a kick answers in a thread of its own and its model context starts empty every time. Without this the agent answers a chat it cannot see; with all of it, the context grows with the chat room's tenure, which is the thing the sub-thread exists to stop.

Fields:

Source code in octomate/config/channels.py
class ChatRecapConfig(BaseModel):
    """What a kick in a dm or a group chat is shown of the chat it woke in.

    Those surfaces have no end, so a kick answers in a thread of its own and its
    model context starts empty every time. Without this the agent answers a chat it
    cannot see; with all of it, the context grows with the chat room's tenure, which
    is the thing the sub-thread exists to stop.
    """

    messages: int = Field(
        default=16,
        ge=0,
        description=(
            "How many of the chat's recent messages go in front of the prompt. "
            "0 shows none, which leaves a kick with only what woke it."
        ),
    )
    characters: int = Field(
        default=1000,
        ge=0,
        description=(
            "Where each of those messages is cut. They are context, not the thing "
            "being answered, and one pasted log would otherwise be the whole slice. "
            "0 leaves them whole."
        ),
    )

messages pydantic-field

messages: int = 16

How many of the chat's recent messages go in front of the prompt. 0 shows none, which leaves a kick with only what woke it.

characters pydantic-field

characters: int = 1000

Where each of those messages is cut. They are context, not the thing being answered, and one pasted log would otherwise be the whole slice. 0 leaves them whole.

ChannelConfig pydantic-model

Bases: BaseModel

What every channel tentacle's block declares: its type, the agents bound to it, and how it streams and recaps.

Fields:

Source code in octomate/config/channels.py
class ChannelConfig(BaseModel):
    """What every channel tentacle's block declares: its type, the agents bound to
    it, and how it streams and recaps."""

    type: str
    mention_only: bool = True
    enabled: bool = True
    stream: ChannelStreamConfig = Field(default_factory=ChannelStreamConfig)
    recap: ChatRecapConfig = Field(default_factory=ChatRecapConfig)
    agents: list[str] = Field(
        min_length=1,
        description=(
            "Bound agent ids, in entry order. The first agent handles new "
            "conversations using its native default; every bound agent exposes "
            "its full model catalog."
        ),
    )
    mcp: bool = Field(
        default=False,
        description=(
            "Whether this channel offers its own MCP tools to the agents driven on "
            "it. The server is one per channel type, so the flag decides what a "
            "call may do here, never which tools an agent sees."
        ),
    )

type pydantic-field

type: str

mention_only pydantic-field

mention_only: bool = True

enabled pydantic-field

enabled: bool = True

stream pydantic-field

stream: ChannelStreamConfig

recap pydantic-field

recap: ChatRecapConfig

agents pydantic-field

agents: list[str]

Bound agent ids, in entry order. The first agent handles new conversations using its native default; every bound agent exposes its full model catalog.

mcp pydantic-field

mcp: bool = False

Whether this channel offers its own MCP tools to the agents driven on it. The server is one per channel type, so the flag decides what a call may do here, never which tools an agent sees.

SlackOAuthClientConfig pydantic-model

Bases: BaseModel

The Slack app as the OAuth client its MCP server takes user tokens from.

Slack's MCP server acts as a person, never as the bot, so a channel that offers it has each user authorize once; the app's client id and secret are what make that authorization a confidential one. The app is the one app_id names, with the "Slack Model Context Protocol (MCP) Server" feature switched on under Agents in its settings.

Fields:

Source code in octomate/config/channels.py
class SlackOAuthClientConfig(BaseModel):
    """The Slack app as the OAuth client its MCP server takes user tokens from.

    Slack's MCP server acts as a person, never as the bot, so a channel that offers
    it has each user authorize once; the app's client id and secret are what make
    that authorization a confidential one. The app is the one `app_id` names, with
    the "Slack Model Context Protocol (MCP) Server" feature switched on under
    Agents in its settings.
    """

    client_id: str = Field(
        description="The app's client id, from Basic Information → App Credentials."
    )
    client_secret: SecretStr = Field(
        description="The app's client secret, from the same page. Prefer the "
        "environment over YAML."
    )
    callback_base_uri: AnyHttpUrl = Field(
        default=AnyHttpUrl("http://localhost:8000"),
        description="Where the authorizing BROWSER reaches this deployment — not "
        "where the internet does, since Slack only redirects the user agent and "
        "never connects here. `<this>/oauth/<channel key>/callback` is the redirect "
        "URL the app must register under OAuth & Permissions, character for "
        "character.",
    )
    scopes: list[SlackUserScope] = Field(
        default_factory=lambda: list(SLACK_MCP_SCOPES),
        description="The user-token scopes each person is asked for when they "
        "connect; the app lists the same under User Token Scopes. Fixed at "
        "authorization: widening this later means every connected user reconnects.",
    )

client_id pydantic-field

client_id: str

The app's client id, from Basic Information → App Credentials.

client_secret pydantic-field

client_secret: SecretStr

The app's client secret, from the same page. Prefer the environment over YAML.

callback_base_uri pydantic-field

callback_base_uri: AnyHttpUrl = AnyHttpUrl(
    "http://localhost:8000"
)

Where the authorizing BROWSER reaches this deployment — not where the internet does, since Slack only redirects the user agent and never connects here. <this>/oauth/<channel key>/callback is the redirect URL the app must register under OAuth & Permissions, character for character.

scopes pydantic-field

scopes: list[SlackUserScope]

The user-token scopes each person is asked for when they connect; the app lists the same under User Token Scopes. Fixed at authorization: widening this later means every connected user reconnects.

SlackChannelConfig pydantic-model

Bases: ChannelConfig

A Slack app, selected by type: slack.

Fields:

Validators:

Source code in octomate/config/channels.py
class SlackChannelConfig(ChannelConfig):
    """A Slack app, selected by `type: slack`."""

    type: Literal["slack"] = "slack"
    app_id: str
    bot_token: SecretStr
    app_token: SecretStr
    stream: SlackStreamConfig = Field(default_factory=SlackStreamConfig)
    oauth: SlackOAuthClientConfig | None = Field(
        default=None,
        description="The app as an OAuth client, which its own MCP tools need: "
        "they act as the person who authorized them, so every user connects once. "
        "Required when `mcp` is on.",
    )

    @model_validator(mode="after")
    def validate_mcp_has_a_client(self) -> Self:
        if self.mcp and self.oauth is None:
            raise ValueError(
                "a Slack channel with `mcp: true` needs an `oauth` block: its MCP "
                "tools act as the person who authorized them, and the app's client "
                "id and secret are what that authorization takes"
            )
        return self

type pydantic-field

type: Literal['slack'] = 'slack'

app_id pydantic-field

app_id: str

bot_token pydantic-field

bot_token: SecretStr

app_token pydantic-field

app_token: SecretStr

stream pydantic-field

stream: SlackStreamConfig

oauth pydantic-field

oauth: SlackOAuthClientConfig | None = None

The app as an OAuth client, which its own MCP tools need: they act as the person who authorized them, so every user connects once. Required when mcp is on.

validate_mcp_has_a_client pydantic-validator

validate_mcp_has_a_client() -> Self
Source code in octomate/config/channels.py
@model_validator(mode="after")
def validate_mcp_has_a_client(self) -> Self:
    if self.mcp and self.oauth is None:
        raise ValueError(
            "a Slack channel with `mcp: true` needs an `oauth` block: its MCP "
            "tools act as the person who authorized them, and the app's client "
            "id and secret are what that authorization takes"
        )
    return self

LarkOAuthClientConfig pydantic-model

Bases: BaseModel

User authorization through the channel's existing Feishu app credentials.

Fields:

Source code in octomate/config/channels.py
class LarkOAuthClientConfig(BaseModel):
    """User authorization through the channel's existing Feishu app credentials."""

    callback_base_uri: AnyHttpUrl = Field(
        default=AnyHttpUrl("http://localhost:8000"),
        description="Browser-reachable origin. Register <origin>/oauth/<channel key>/callback "
        "under the app's Security Settings → Redirect URLs.",
    )
    scopes: list[str] = Field(
        default_factory=lambda: [
            "offline_access",
            "contact:contact.base:readonly",
            "contact:user.base:readonly",
        ],
        description="User permissions requested during authorization. Enable these in the "
        "app's User Identity Permissions; reconnect after widening them. The defaults "
        "allow get-user and token refresh.",
    )

callback_base_uri pydantic-field

callback_base_uri: AnyHttpUrl = AnyHttpUrl(
    "http://localhost:8000"
)

Browser-reachable origin. Register /oauth//callback under the app's Security Settings → Redirect URLs.

scopes pydantic-field

scopes: list[str]

User permissions requested during authorization. Enable these in the app's User Identity Permissions; reconnect after widening them. The defaults allow get-user and token refresh.

LarkChannelConfig pydantic-model

Bases: ChannelConfig

A Lark app, selected by type: lark.

Fields:

Validators:

Source code in octomate/config/channels.py
class LarkChannelConfig(ChannelConfig):
    """A Lark app, selected by `type: lark`."""

    type: Literal["lark"] = "lark"
    app_id: str
    app_secret: SecretStr
    stream: LarkStreamConfig = Field(default_factory=LarkStreamConfig)
    oauth: LarkOAuthClientConfig | None = Field(
        default=None,
        description="User OAuth settings, using this channel's app_id and app_secret. "
        "Required when mcp is enabled; also allows profile linking from Trunkline.",
    )
    mcp_tools: list[str] = Field(
        default_factory=lambda: ["get-user"],
        min_length=1,
        description="Feishu remote MCP tool names sent in X-Lark-MCP-Allowed-Tools. "
        "Their required user permissions must also be listed in oauth.scopes.",
    )

    @model_validator(mode="after")
    def validate_mcp_has_a_client(self) -> Self:
        if self.mcp and self.oauth is None:
            raise ValueError("a Lark channel with `mcp: true` needs an `oauth` block")
        return self

type pydantic-field

type: Literal['lark'] = 'lark'

app_id pydantic-field

app_id: str

app_secret pydantic-field

app_secret: SecretStr

stream pydantic-field

stream: LarkStreamConfig

oauth pydantic-field

oauth: LarkOAuthClientConfig | None = None

User OAuth settings, using this channel's app_id and app_secret. Required when mcp is enabled; also allows profile linking from Trunkline.

mcp_tools pydantic-field

mcp_tools: list[str]

Feishu remote MCP tool names sent in X-Lark-MCP-Allowed-Tools. Their required user permissions must also be listed in oauth.scopes.

validate_mcp_has_a_client pydantic-validator

validate_mcp_has_a_client() -> Self
Source code in octomate/config/channels.py
@model_validator(mode="after")
def validate_mcp_has_a_client(self) -> Self:
    if self.mcp and self.oauth is None:
        raise ValueError("a Lark channel with `mcp: true` needs an `oauth` block")
    return self

DiscordOAuthClientConfig pydantic-model

Bases: BaseModel

The Discord application's OAuth2 client, used to link profiles by identity.

Fields:

Source code in octomate/config/channels.py
class DiscordOAuthClientConfig(BaseModel):
    """The Discord application's OAuth2 client, used to link profiles by identity."""

    client_id: str = Field(
        min_length=1, description="The Discord application's OAuth2 client id."
    )
    client_secret: SecretStr = Field(
        min_length=1,
        description="The application's OAuth2 client secret, not its bot token.",
    )

client_id pydantic-field

client_id: str

The Discord application's OAuth2 client id.

client_secret pydantic-field

client_secret: SecretStr

The application's OAuth2 client secret, not its bot token.

DiscordChannelConfig pydantic-model

Bases: ChannelConfig

A Discord bot, selected by type: discord.

Fields:

Source code in octomate/config/channels.py
class DiscordChannelConfig(ChannelConfig):
    """A Discord bot, selected by `type: discord`."""

    type: Literal["discord"] = "discord"
    bot_token: SecretStr
    stream: DiscordStreamConfig = Field(default_factory=DiscordStreamConfig)
    oauth: DiscordOAuthClientConfig | None = Field(
        default=None,
        description="Optional user authorization for profile linking; requests identify only.",
    )

type pydantic-field

type: Literal['discord'] = 'discord'

bot_token pydantic-field

bot_token: SecretStr

stream pydantic-field

stream: DiscordStreamConfig

oauth pydantic-field

oauth: DiscordOAuthClientConfig | None = None

Optional user authorization for profile linking; requests identify only.

TrunklineStreamConfig pydantic-model

Bases: ChannelStreamConfig

Streaming defaults for the console: every event straight through.

Fields:

Source code in octomate/config/channels.py
class TrunklineStreamConfig(ChannelStreamConfig):
    """Streaming defaults for the console: every event straight through."""

    # The console renders tokens as it receives them; stream every event
    # straight through (the timeline feeler forwards raw events, so batching
    # is moot). The stream is the console's only delivery, so it cannot be off.
    enabled: Literal[True] = True
    flush_interval: float = 0.0

enabled pydantic-field

enabled: Literal[True] = True

flush_interval pydantic-field

flush_interval: float = 0.0

TrunklineChannelConfig pydantic-model

Bases: ChannelConfig

The console has no group surface, so the base mention_only never applies to it (there is nothing to be mentioned in). Configured agents set entry order; every registered agent's models are available in the console.

Fields:

Source code in octomate/config/channels.py
class TrunklineChannelConfig(ChannelConfig):
    """The console has no group surface, so the base `mention_only` never
    applies to it (there is nothing to be mentioned in). Configured agents set
    entry order; every registered agent's models are available in the console."""

    type: Literal["trunkline"] = "trunkline"
    static_dir: ConfigPath | None = Field(
        default=None,
        description=(
            "Directory containing the compiled UI, served at /. Unset disables "
            "static serving without disabling the console API. Expands ~; relative "
            "paths use the server's working directory. The directory must exist."
        ),
    )
    # Annotated as the subclass so a YAML `stream:` override keeps the
    # console defaults (the base class would flip `enabled` back to False).
    stream: TrunklineStreamConfig = Field(default_factory=TrunklineStreamConfig)

type pydantic-field

type: Literal['trunkline'] = 'trunkline'

static_dir pydantic-field

static_dir: ConfigPath | None = None

Directory containing the compiled UI, served at /. Unset disables static serving without disabling the console API. Expands ~; relative paths use the server's working directory. The directory must exist.

stream pydantic-field

stream: TrunklineStreamConfig