Skip to content

line_channel

line_channel

LineChannel — LINE Messaging API adapter via line-bot-sdk.

Classes

LineChannel 

LineChannel(channel_access_token: str = '', *, channel_secret: str = '', bus: Optional[EventBus] = None)

Bases: BaseChannel

LINE messaging channel adapter.

Uses the LINE Messaging API via line-bot-sdk.

PARAMETER DESCRIPTION
channel_access_token

LINE channel access token. Falls back to LINE_CHANNEL_ACCESS_TOKEN env var.

TYPE: str DEFAULT: ''

channel_secret

LINE channel secret. Falls back to LINE_CHANNEL_SECRET env var.

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/openjarvis/channels/line_channel.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    channel_access_token: str = "",
    *,
    channel_secret: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._channel_access_token = channel_access_token or os.environ.get(
        "LINE_CHANNEL_ACCESS_TOKEN", ""
    )
    self._channel_secret = channel_secret or os.environ.get(
        "LINE_CHANNEL_SECRET", ""
    )
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
Functions
connect 
connect() -> None

Validate credentials and mark as connected.

Source code in src/openjarvis/channels/line_channel.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._channel_access_token or not self._channel_secret:
        logger.warning("No LINE channel_access_token or channel_secret configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import linebot  # noqa: F401
    except ImportError:
        raise ImportError(
            "line-bot-sdk not installed. Install with: "
            "uv sync --extra channel-line"
        )
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/openjarvis/channels/line_channel.py 
73
74
75
def disconnect(self) -> None:
    """Mark as disconnected."""
    self._status = ChannelStatus.DISCONNECTED
send 
send(channel: str, content: str, *, conversation_id: str = '', metadata: Dict[str, Any] | None = None) -> bool

Send a push message to a LINE user or group.

PARAMETER DESCRIPTION
channel

LINE user ID or group ID to send to.

TYPE: str

content

Text message content.

TYPE: str

Source code in src/openjarvis/channels/line_channel.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a push message to a LINE user or group.

    Parameters
    ----------
    channel:
        LINE user ID or group ID to send to.
    content:
        Text message content.
    """
    if not self._channel_access_token:
        logger.warning("Cannot send: no LINE credentials configured")
        return False

    try:
        from linebot.v3.messaging import (
            ApiClient,
            Configuration,
            MessagingApi,
            PushMessageRequest,
            TextMessage,
        )

        configuration = Configuration(
            access_token=self._channel_access_token,
        )
        with ApiClient(configuration) as api_client:
            api = MessagingApi(api_client)
            api.push_message(
                PushMessageRequest(
                    to=channel,
                    messages=[TextMessage(text=content)],
                )
            )

        self._publish_sent(channel, content, conversation_id)
        return True
    except ImportError:
        logger.debug("line-bot-sdk not installed")
        return False
    except Exception:
        logger.debug("LINE send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/openjarvis/channels/line_channel.py 
130
131
132
def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status
list_channels 
list_channels() -> List[str]

Return available channel identifiers.

Source code in src/openjarvis/channels/line_channel.py 
134
135
136
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["line"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/openjarvis/channels/line_channel.py 
138
139
140
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)