feat(msteams): wire agent integration for Teams messages

- Integrate dispatchReplyFromConfig() for full agent routing
- Add msteams to TextChunkProvider and OriginatingChannelType
- Add msteams case to route-reply (proactive not yet supported)
- Strip @mention HTML tags from Teams messages
- Fix session key to exclude messageid suffix
- Add typing indicator support
- Add proper logging for debugging
This commit is contained in:
Onur
2026-01-07 23:00:59 +03:00
committed by Peter Steinberger
parent 951789e9fa
commit 1c73d45106
5 changed files with 328 additions and 34 deletions

View File

@@ -17,7 +17,8 @@ export type TextChunkProvider =
| "slack"
| "signal"
| "imessage"
| "webchat";
| "webchat"
| "msteams";
const DEFAULT_CHUNK_LIMIT_BY_PROVIDER: Record<TextChunkProvider, number> = {
whatsapp: 4000,
@@ -27,6 +28,7 @@ const DEFAULT_CHUNK_LIMIT_BY_PROVIDER: Record<TextChunkProvider, number> = {
signal: 4000,
imessage: 4000,
webchat: 4000,
msteams: 4000,
};
export function resolveTextChunkLimit(

View File

@@ -145,6 +145,14 @@ export async function routeReply(
};
}
case "msteams": {
// TODO: Implement proactive messaging for MS Teams
return {
ok: false,
error: `MS Teams routing not yet supported for queued replies`,
};
}
default: {
const _exhaustive: never = channel;
return { ok: false, error: `Unknown channel: ${String(_exhaustive)}` };

View File

@@ -6,7 +6,8 @@ export type OriginatingChannelType =
| "signal"
| "imessage"
| "whatsapp"
| "webchat";
| "webchat"
| "msteams";
export type MsgContext = {
Body?: string;

View File

@@ -1,9 +1,21 @@
import {
chunkMarkdownText,
resolveTextChunkLimit,
} from "../auto-reply/chunk.js";
import { formatAgentEnvelope } from "../auto-reply/envelope.js";
import { dispatchReplyFromConfig } from "../auto-reply/reply/dispatch-from-config.js";
import { createReplyDispatcherWithTyping } from "../auto-reply/reply/reply-dispatcher.js";
import { SILENT_REPLY_TOKEN } from "../auto-reply/tokens.js";
import type { ReplyPayload } from "../auto-reply/types.js";
import type { ClawdbotConfig } from "../config/types.js";
import { danger, logVerbose, shouldLogVerbose } from "../globals.js";
import { enqueueSystemEvent } from "../infra/system-events.js";
import { getChildLogger } from "../logging.js";
import { resolveAgentRoute } from "../routing/resolve-route.js";
import type { RuntimeEnv } from "../runtime.js";
import { resolveMSTeamsCredentials } from "./token.js";
const log = getChildLogger({ name: "msteams:monitor" });
const log = getChildLogger({ name: "msteams" });
export type MonitorMSTeamsOpts = {
cfg: ClawdbotConfig;
@@ -16,10 +28,45 @@ export type MonitorMSTeamsResult = {
shutdown: () => Promise<void>;
};
type TeamsActivity = {
id?: string;
type?: string;
timestamp?: string | Date;
text?: string;
from?: { id?: string; name?: string; aadObjectId?: string };
recipient?: { id?: string; name?: string };
conversation?: {
id?: string;
conversationType?: string;
tenantId?: string;
isGroup?: boolean;
};
channelId?: string;
serviceUrl?: string;
membersAdded?: Array<{ id?: string; name?: string }>;
};
type TeamsTurnContext = {
activity: TeamsActivity;
sendActivity: (textOrActivity: string | object) => Promise<unknown>;
sendActivities?: (
activities: Array<{ type: string } & Record<string, unknown>>,
) => Promise<unknown>;
};
// Helper to convert timestamp to Date
function parseTimestamp(ts?: string | Date): Date | undefined {
if (!ts) return undefined;
if (ts instanceof Date) return ts;
const date = new Date(ts);
return Number.isNaN(date.getTime()) ? undefined : date;
}
export async function monitorMSTeamsProvider(
opts: MonitorMSTeamsOpts,
): Promise<MonitorMSTeamsResult> {
const msteamsCfg = opts.cfg.msteams;
const cfg = opts.cfg;
const msteamsCfg = cfg.msteams;
if (!msteamsCfg?.enabled) {
log.debug("msteams provider disabled");
return { app: null, shutdown: async () => {} };
@@ -31,46 +78,246 @@ export async function monitorMSTeamsProvider(
return { app: null, shutdown: async () => {} };
}
const port = msteamsCfg.webhook?.port ?? 3978;
const path = msteamsCfg.webhook?.path ?? "/msteams/messages";
const runtime: RuntimeEnv = opts.runtime ?? {
log: console.log,
error: console.error,
exit: (code: number): never => {
throw new Error(`exit ${code}`);
},
};
log.info(`starting msteams provider on port ${port}${path}`);
const port = msteamsCfg.webhook?.port ?? 3978;
const textLimit = resolveTextChunkLimit(cfg, "msteams");
log.info(`starting provider (port ${port})`);
// Dynamic import to avoid loading SDK when provider is disabled
const agentsHosting = await import("@microsoft/agents-hosting");
const { startServer } = await import("@microsoft/agents-hosting-express");
const { ActivityHandler } = agentsHosting;
type TurnContext = InstanceType<typeof agentsHosting.TurnContext>;
// Create activity handler using fluent API
const handler = new ActivityHandler()
.onMessage(async (context: TurnContext, next: () => Promise<void>) => {
const text = context.activity?.text?.trim() ?? "";
const from = context.activity?.from;
const conversation = context.activity?.conversation;
// Helper to deliver replies via Teams SDK
async function deliverReplies(params: {
replies: ReplyPayload[];
context: TeamsTurnContext;
}) {
const chunkLimit = Math.min(textLimit, 4000);
for (const payload of params.replies) {
const mediaList =
payload.mediaUrls ?? (payload.mediaUrl ? [payload.mediaUrl] : []);
const text = payload.text ?? "";
if (!text && mediaList.length === 0) continue;
log.debug("received message", {
text: text.slice(0, 100),
from: from?.id,
conversation: conversation?.id,
if (mediaList.length === 0) {
for (const chunk of chunkMarkdownText(text, chunkLimit)) {
const trimmed = chunk.trim();
if (!trimmed || trimmed === SILENT_REPLY_TOKEN) continue;
await params.context.sendActivity(trimmed);
}
} else {
// For media, send text first then media URLs as separate messages
if (text.trim() && text.trim() !== SILENT_REPLY_TOKEN) {
for (const chunk of chunkMarkdownText(text, chunkLimit)) {
await params.context.sendActivity(chunk);
}
}
for (const mediaUrl of mediaList) {
// Teams supports adaptive cards for rich media, but for now just send URL
await params.context.sendActivity(mediaUrl);
}
}
}
}
// Strip Teams @mention HTML tags from message text
function stripMentionTags(text: string): string {
// Teams wraps mentions in <at>...</at> tags
return text.replace(/<at>.*?<\/at>/gi, "").trim();
}
// Handler for incoming messages
async function handleTeamsMessage(context: TeamsTurnContext) {
const activity = context.activity;
const rawText = activity.text?.trim() ?? "";
const text = stripMentionTags(rawText);
const from = activity.from;
const conversation = activity.conversation;
log.info("received message", {
rawText: rawText.slice(0, 50),
text: text.slice(0, 50),
from: from?.id,
conversation: conversation?.id,
});
if (!text) {
log.debug("skipping empty message after stripping mentions");
return;
}
if (!from?.id) {
log.debug("skipping message without from.id");
return;
}
// Teams conversation.id may include ";messageid=..." suffix - strip it for session key
const rawConversationId = conversation?.id ?? "";
const conversationId = rawConversationId.split(";")[0];
const conversationType = conversation?.conversationType ?? "personal";
const isGroupChat =
conversationType === "groupChat" || conversation?.isGroup === true;
const isChannel = conversationType === "channel";
const isDirectMessage = !isGroupChat && !isChannel;
const senderName = from.name ?? from.id;
const senderId = from.aadObjectId ?? from.id;
// Build Teams-specific identifiers
const teamsFrom = isDirectMessage
? `msteams:${senderId}`
: isChannel
? `msteams:channel:${conversationId}`
: `msteams:group:${conversationId}`;
const teamsTo = isDirectMessage
? `user:${senderId}`
: `conversation:${conversationId}`;
// Resolve routing
const route = resolveAgentRoute({
cfg,
provider: "msteams",
peer: {
kind: isDirectMessage ? "dm" : isChannel ? "channel" : "group",
id: isDirectMessage ? senderId : conversationId,
},
});
const preview = text.replace(/\s+/g, " ").slice(0, 160);
const inboundLabel = isDirectMessage
? `Teams DM from ${senderName}`
: `Teams message in ${conversationType} from ${senderName}`;
enqueueSystemEvent(`${inboundLabel}: ${preview}`, {
sessionKey: route.sessionKey,
contextKey: `msteams:message:${conversationId}:${activity.id ?? "unknown"}`,
});
// Format the message body with envelope
const timestamp = parseTimestamp(activity.timestamp);
const body = formatAgentEnvelope({
provider: "Teams",
from: senderName,
timestamp,
body: text,
});
// Build context payload for agent
const ctxPayload = {
Body: body,
From: teamsFrom,
To: teamsTo,
SessionKey: route.sessionKey,
AccountId: route.accountId,
ChatType: isDirectMessage ? "direct" : isChannel ? "room" : "group",
GroupSubject: !isDirectMessage ? conversationType : undefined,
SenderName: senderName,
SenderId: senderId,
Provider: "msteams" as const,
Surface: "msteams" as const,
MessageSid: activity.id,
Timestamp: timestamp?.getTime() ?? Date.now(),
WasMentioned: !isDirectMessage,
CommandAuthorized: true,
OriginatingChannel: "msteams" as const,
OriginatingTo: teamsTo,
};
if (shouldLogVerbose()) {
logVerbose(
`msteams inbound: from=${ctxPayload.From} preview="${preview}"`,
);
}
// Send typing indicator
const sendTypingIndicator = async () => {
try {
if (context.sendActivities) {
await context.sendActivities([{ type: "typing" }]);
}
} catch {
// Typing indicator is best-effort
}
};
// Create reply dispatcher
const { dispatcher, replyOptions, markDispatchIdle } =
createReplyDispatcherWithTyping({
responsePrefix: cfg.messages?.responsePrefix,
deliver: async (payload) => {
await deliverReplies({
replies: [payload],
context,
});
},
onError: (err, info) => {
runtime.error?.(
danger(`msteams ${info.kind} reply failed: ${String(err)}`),
);
},
onReplyStart: sendTypingIndicator,
});
// TODO: Implement full message handling
// - Route to agent based on config
// - Process commands
// - Send reply via context.sendActivity()
// Dispatch to agent
log.info("dispatching to agent", { sessionKey: route.sessionKey });
try {
const { queuedFinal, counts } = await dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions,
});
// Echo for now as a test
await context.sendActivity(`Received: ${text}`);
markDispatchIdle();
log.info("dispatch complete", { queuedFinal, counts });
if (!queuedFinal) return;
if (shouldLogVerbose()) {
const finalCount = counts.final;
logVerbose(
`msteams: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${teamsTo}`,
);
}
} catch (err) {
log.error("dispatch failed", { error: String(err) });
runtime.error?.(danger(`msteams dispatch failed: ${String(err)}`));
// Try to send error message back to Teams
try {
await context.sendActivity(
`⚠️ Agent failed: ${err instanceof Error ? err.message : String(err)}`,
);
} catch {
// Best effort
}
}
}
// Create activity handler using fluent API
// The SDK's TurnContext is compatible with our TeamsTurnContext
const handler = new ActivityHandler()
.onMessage(async (context, next) => {
try {
await handleTeamsMessage(context as unknown as TeamsTurnContext);
} catch (err) {
runtime.error?.(danger(`msteams handler failed: ${String(err)}`));
}
await next();
})
.onMembersAdded(async (context: TurnContext, next: () => Promise<void>) => {
.onMembersAdded(async (context, next) => {
const membersAdded = context.activity?.membersAdded ?? [];
for (const member of membersAdded) {
if (member.id !== context.activity?.recipient?.id) {
log.debug("member added", { member: member.id });
await context.sendActivity("Hello! I'm Clawdbot.");
// Don't send welcome message - let the user initiate conversation
}
}
await next();

View File

@@ -831,14 +831,50 @@ Initial recommendation: support this type first; treat other attachment types as
2.**Config plumbing**: `MSTeamsConfig` type + zod schema (`src/config/types.ts`, `src/config/zod-schema.ts`)
3.**Provider skeleton**: `src/msteams/` with `index.ts`, `token.ts`, `probe.ts`, `send.ts`, `monitor.ts`
4.**Gateway integration**: Provider manager start/stop wiring in `server-providers.ts` and `server.ts`
5.**Echo bot tested**: Verified end-to-end flow (Azure Bot → Tailscale → Gateway → SDK → Response)
### Debugging Notes
- **SDK listens on all paths**: The `startServer()` function responds to POST on any path (not just `/api/messages`), but Azure Bot default is `/api/messages`
- **SDK handles HTTP internally**: Custom logging in monitor.ts `log.debug()` doesn't show HTTP traffic - SDK processes requests before our handler
- **Tailscale Funnel**: Must be running separately (`tailscale funnel 3978`) - doesn't work well as background task
- **Auth errors (401)**: Expected when testing manually without Azure JWT - means endpoint is reachable
### In Progress (2026-01-07 - Session 2)
6.**Agent dispatch (sync)**: Wired inbound messages to `dispatchReplyFromConfig()` - replies sent via `context.sendActivity()` within turn
7.**Typing indicator**: Added typing indicator support via `sendActivities([{ type: "typing" }])`
8.**Type system updates**: Added `msteams` to `TextChunkProvider`, `OriginatingChannelType`, and route-reply switch
### Implementation Notes
**Current Approach (Synchronous):**
The current implementation sends replies synchronously within the Teams turn context. This works for quick responses but may timeout for slow LLM responses.
```typescript
// Current: Reply within turn context (src/msteams/monitor.ts)
const { dispatcher, replyOptions, markDispatchIdle } = createReplyDispatcherWithTyping({
deliver: async (payload) => {
await deliverReplies({ replies: [payload], context });
},
onReplyStart: sendTypingIndicator,
});
await dispatchReplyFromConfig({ ctx: ctxPayload, cfg, dispatcher, replyOptions });
```
**Key Fields in ctxPayload:**
- `Provider: "msteams"` / `Surface: "msteams"`
- `From`: `msteams:<userId>` (DM) or `msteams:channel:<conversationId>` (channel)
- `To`: `user:<userId>` (DM) or `conversation:<conversationId>` (group/channel)
- `ChatType`: `"direct"` | `"group"` | `"room"` based on conversation type
### Remaining
5. **Test echo bot**: Run gateway with msteams enabled, verify Teams can reach the webhook and receive echo replies.
6. **Conversation store**: Persist `ConversationReference` by `conversation.id` for proactive messaging.
7. **Agent dispatch (async)**: Wire inbound messages to `dispatchReplyFromConfig()` using proactive sends.
8. **Access control**: Implement DM policy + pairing (reuse existing pairing store) + mention gating in channels.
9. **Config reload**: Add msteams to `config-reload.ts` restart rules.
10. **Outbound CLI/gateway sends**: Implement `sendMessageMSTeams` properly; wire `clawdbot send --provider msteams`.
11. **Media**: Implement inbound attachment download and outbound strategy.
12. **Docs + UI + Onboard**: Write `docs/providers/msteams.md`, add UI config form, update `clawdbot onboard`.
9. **Test full agent flow**: Send message in Teams → verify agent responds (not just echo)
10. **Conversation store**: Persist `ConversationReference` by `conversation.id` for proactive messaging
11. **Proactive messaging**: For slow LLM responses, store reference and send replies asynchronously
12. **Access control**: Implement DM policy + pairing (reuse existing pairing store) + mention gating in channels
13. **Config reload**: Add msteams to `config-reload.ts` restart rules
14. **Outbound CLI/gateway sends**: Implement `sendMessageMSTeams` properly; wire `clawdbot send --provider msteams`
15. **Media**: Implement inbound attachment download and outbound strategy
16. **Docs + UI + Onboard**: Write `docs/providers/msteams.md`, add UI config form, update `clawdbot onboard`