Skip to content

teams

teams

TeamsChannel — Microsoft Teams Bot Framework adapter.

Classes

TeamsChannel 

TeamsChannel(app_id: str = '', *, app_password: str = '', service_url: str = '', bus: Optional[EventBus] = None)

Bases: BaseChannel

Microsoft Teams channel adapter using the Bot Framework REST API.

PARAMETER DESCRIPTION
app_id

Microsoft App ID. Falls back to TEAMS_APP_ID env var.

TYPE: str DEFAULT: ''

app_password

Microsoft App Password. Falls back to TEAMS_APP_PASSWORD env var.

TYPE: str DEFAULT: ''

service_url

Bot Framework service URL. Falls back to TEAMS_SERVICE_URL env var (default https://smba.trafficmanager.net/teams).

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/openjarvis/channels/teams.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    app_id: str = "",
    *,
    app_password: str = "",
    service_url: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._app_id = app_id or os.environ.get("TEAMS_APP_ID", "")
    self._app_password = app_password or os.environ.get("TEAMS_APP_PASSWORD", "")
    self._service_url = (
        service_url
        or os.environ.get("TEAMS_SERVICE_URL", "https://smba.trafficmanager.net/teams")
    )
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
Functions
connect 
connect() -> None

Mark as connected (send-only — no persistent connection).

Source code in src/openjarvis/channels/teams.py 
59
60
61
62
63
64
65
def connect(self) -> None:
    """Mark as connected (send-only — no persistent connection)."""
    if not self._app_id or not self._app_password:
        logger.warning("No Teams app_id or app_password configured")
        self._status = ChannelStatus.ERROR
        return
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/openjarvis/channels/teams.py 
67
68
69
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 message to a Teams conversation via the Bot Framework API.

Source code in src/openjarvis/channels/teams.py 
 73
 74
 75
 76
 77
 78
 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
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Teams conversation via the Bot Framework API."""
    if not self._app_id or not self._app_password:
        logger.warning("Cannot send: no Teams credentials configured")
        return False

    try:
        import httpx

        url = f"{self._service_url}/v3/conversations/{channel}/activities"
        headers = {
            "Authorization": f"Bearer {self._app_password}",
            "Content-Type": "application/json",
        }
        payload: Dict[str, Any] = {
            "type": "message",
            "text": content,
        }
        if conversation_id:
            payload["replyToId"] = conversation_id

        resp = httpx.post(
            url, json=payload, headers=headers, timeout=10.0,
        )
        if resp.status_code < 300:
            self._publish_sent(channel, content, conversation_id)
            return True
        logger.warning(
            "Teams API returned status %d", resp.status_code,
        )
        return False
    except Exception:
        logger.debug("Teams send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/openjarvis/channels/teams.py 
115
116
117
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/teams.py 
119
120
121
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["teams"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/openjarvis/channels/teams.py 
123
124
125
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)