Skip to content

channel_agent

channel_agent

ChannelAgent — bridge between messaging channels and AI agents.

Routes incoming :class:~openjarvis.channels._stubs.ChannelMessage objects to an agent, classifies queries as "quick" or "deep", and delivers responses either inline or as a preview with an escalation link to a full report.

Classes

ChannelAgent 

ChannelAgent(channel: BaseChannel, agent: Any, *, max_workers: int = 2)

Bridge between a :class:BaseChannel and an agent.

On each incoming message the agent is invoked in a background thread so that :meth:_handle_message never blocks the channel's event loop. Quick queries with short responses are delivered inline; deep queries or long responses trigger a preview + escalation link.

PARAMETER DESCRIPTION
channel

A connected :class:BaseChannel instance.

TYPE: BaseChannel

agent

Any object that exposes a run(input: str) -> AgentResult-compatible method (typically a :class:~openjarvis.agents._stubs.BaseAgent subclass).

TYPE: Any

max_workers

Size of the background :class:~concurrent.futures.ThreadPoolExecutor.

TYPE: int DEFAULT: 2

Source code in src/openjarvis/agents/channel_agent.py 
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    channel: BaseChannel,
    agent: Any,
    *,
    max_workers: int = 2,
) -> None:
    self._channel = channel
    self._agent = agent
    self._pool = ThreadPoolExecutor(max_workers=max_workers)
    channel.on_message(self._handle_message)
Functions
shutdown 
shutdown() -> None

Shut down the background thread pool.

Source code in src/openjarvis/agents/channel_agent.py 
143
144
145
def shutdown(self) -> None:
    """Shut down the background thread pool."""
    self._pool.shutdown(wait=True)

Functions

classify_query 

classify_query(text: str) -> str

Return 'quick' or 'deep' based on heuristics.

Deep signals take priority over quick signals. A query is classified as deep if it contains deep keywords, a time-range phrase, or is longer than 20 words. Otherwise it is quick.

Source code in src/openjarvis/agents/channel_agent.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def classify_query(text: str) -> str:
    """Return ``'quick'`` or ``'deep'`` based on heuristics.

    Deep signals take priority over quick signals.  A query is classified as
    deep if it contains deep keywords, a time-range phrase, or is longer than
    20 words.  Otherwise it is quick.
    """
    words = text.split()
    if _DEEP_KEYWORDS.search(text):
        return "deep"
    if _TIME_RANGE.search(text):
        return "deep"
    if len(words) > 20:
        return "deep"
    return "quick"