merge: resolve conflicts with main (Part B + voice-agent merged)

Reconcile meeting-agent Phase A with main after #266/#267: keep
meeting_integrations/ package rename, merge Part B live-transcript and
Calendar V2 routes, and preserve Part A fixes (atomic webhook idempotency,
background task refs, workspace ID placeholders, py3.10 compiler compat).
Update part_b tests to meeting_integrations paths.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
MervinPraisonandCursor committed 2026-09-08 10:34:47 +01:00
commit e15b3c0718
48 files changed
+2932 -628

No files matched your search

+4 -4
View File
@@ -1,9 +1,9 @@
{
"templateRepo": "MervinPraison/github-automation-template",
"stack": "claude",
"profile": "docs",
"profile": "sdk",
"mode": "hub",
"hubRef": "v1.1.1",
"templateVersion": "1.1.1",
"installedAt": "2026-09-02T12:07:52Z"
"hubRef": "v1.1.6",
"templateVersion": "1.1.6",
"installedAt": "2026-09-07T10:02:11Z"
}
+8 -3
View File
@@ -18,11 +18,16 @@ env:
permissions:
issues: write
pull-requests: read
pull-requests: write
jobs:
review-chain:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
# Grant union of callee job permissions (repos with default_workflow_permissions: read
# reject reusable calls when the caller ceiling is lower than any callee job).
permissions:
issues: write
pull-requests: write
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
secrets: inherit
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
+2 -2
View File
@@ -16,8 +16,8 @@ permissions:
jobs:
recover:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected].1
uses: MervinPraison/github-automation-template/.github/workflows/[email protected].6
secrets: inherit
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
pr_number: ${{ github.event.inputs.pr_number || '' }}
+3 -3
View File
@@ -2,7 +2,7 @@ name: Auto CI Failure → Claude
on:
workflow_run:
workflows: [CI, CI]
workflows: [Core Tests, Optimized Test Suite]
types: [completed]
pull_request:
types: [opened, synchronize, reopened]
@@ -25,8 +25,8 @@ permissions:
jobs:
detect-and-trigger:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected].1
uses: MervinPraison/github-automation-template/.github/workflows/[email protected].6
secrets: inherit
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
pr_number: ${{ github.event.inputs.pr_number || '' }}
+12 -6
View File
@@ -16,7 +16,7 @@ on:
workflow_run:
workflows:
- Claude Assistant
- CI
- Core Tests
types: [completed]
repository_dispatch:
types: [claude-merge-gate]
@@ -43,9 +43,9 @@ jobs:
id-token: write
steps:
- uses: actions/checkout@v4
- uses: MervinPraison/github-automation-template/.github/actions/[email protected].1
- uses: MervinPraison/github-automation-template/.github/actions/[email protected].6
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
- name: Generate GitHub App token
id: app-token
uses: actions/create-github-app-token@v1
@@ -64,7 +64,7 @@ jobs:
const owner = context.repo.owner;
const repo = context.repo.repo;
if (wr.conclusion !== 'success') return;
const allowed = ['Claude Assistant', 'CI'];
const allowed = ['Claude Assistant', 'Core Tests'];
if (!allowed.includes(wr.name)) return;
const prNumber = await mergeGate.resolvePrNumberFromWorkflowRun(github, owner, repo, wr);
if (!prNumber) return;
@@ -81,9 +81,15 @@ jobs:
merge-gate:
if: github.event_name != 'workflow_run'
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
permissions:
contents: write
pull-requests: write
issues: write
actions: read
id-token: write
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
secrets: inherit
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
pr_number: ${{ github.event.inputs.pr_number || github.event.client_payload.pr_number || '' }}
merge_method: ${{ github.event.inputs.merge_method || 'merge' }}
+6 -541
View File
@@ -1,559 +1,24 @@
name: Claude Assistant
# Triggers:
# - issues:opened → lightweight triage (labels + auto-claude for owner)
# - issues:labeled → full implementation when `claude` label applied
# - issues:assigned → full implementation when issue assigned to Claude
# - issue_comment → full implementation when @claude mentioned (issues + PRs)
# - pull_request_review → full implementation when @claude mentioned in review
# - pull_request_review_comment → full implementation when @claude mentioned inline
#
# Escalate an issue with the `claude` label OR an @claude comment — not both.
# If both fire within 3 minutes, the comment trigger is skipped (dedupe job).
#
# Focus: CLAUDE only. Copilot and CodeRabbit are GitHub Apps that react to PRs
# automatically — no workflow file needed. auto-pr-comment.yml orchestrates the
# CodeRabbit → Copilot → Claude review chain for PRs.
on:
issue_comment:
types: [created]
pull_request_review_comment:
types: [created]
issues:
types: [opened, labeled, assigned]
types: [assigned, labeled, opened]
pull_request_review:
types: [submitted]
workflow_dispatch:
inputs:
issue_number:
description: 'Issue number for Claude implementation'
required: true
type: string
jobs:
# ================================================================
# Job 1 — Issue triage (runs on ANY new issue, including external users)
# Posts welcome comment + triage labels. Auto-adds `claude` for owner issues.
# ================================================================
issue-triage:
if: |
github.event_name == 'issues' &&
github.event.action == 'opened' &&
!contains(toJSON(github.event.issue.labels), 'claude')
runs-on: ubuntu-latest
permissions:
issues: write
contents: read
steps:
- name: Generate GitHub App Token
id: app-token
continue-on-error: true
uses: actions/create-github-app-token@v1
with:
app-id: ${{ secrets.CLAUDE_APP_ID }}
private-key: ${{ secrets.CLAUDE_APP_PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
- name: Resolve GitHub token
id: github-token
run: |
if [ -n "${{ steps.app-token.outputs.token }}" ]; then
echo "token=${{ steps.app-token.outputs.token }}" >> "$GITHUB_OUTPUT"
else
echo "token=${{ secrets.GITHUB_TOKEN }}" >> "$GITHUB_OUTPUT"
fi
- name: Triage and acknowledge issue
uses: actions/github-script@v7
with:
github-token: ${{ steps.github-token.outputs.token }}
script: |
const issue = context.payload.issue;
const title = (issue.title || '').toLowerCase();
const body = (issue.body || '').toLowerCase();
const content = title + ' ' + body;
const isOwner = ['OWNER', 'MEMBER', 'COLLABORATOR'].includes(
issue.author_association
);
const labels = [];
if (content.includes('feature request') || content.includes('[feature')) {
labels.push('enhancement');
} else if (content.includes('bug') || content.includes('error') || content.includes('crash') || content.includes('traceback')) {
labels.push('bug');
} else if (content.includes('question') || content.includes('how do i') || content.includes('how to')) {
labels.push('question');
}
if (content.includes('security') || content.includes('cve') || content.includes('vulnerability')) {
labels.push('security');
}
if (content.includes('performance') || content.includes('slow') || content.includes('memory leak')) {
labels.push('performance');
}
if (content.includes('documentation') || content.includes('docs')) {
labels.push('documentation');
}
if (content.includes('typescript') || content.includes('javascript') || content.includes('npm')) {
labels.push('javascript');
}
if (content.includes('frontend') || content.includes('react') || content.includes('vite')) {
labels.push('frontend');
}
if (content.includes('example') || content.includes('yaml')) {
labels.push('examples');
}
if (labels.length > 0) {
await github.rest.issues.addLabels({
issue_number: issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
labels
});
}
if (!isOwner) {
await github.rest.issues.createComment({
issue_number: issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: [
`👋 Thanks for opening this issue, @${issue.user.login}!`,
'',
'A maintainer will review this shortly. In the meantime:',
'- Make sure you\'ve included steps to reproduce (for bugs)',
'- Check [existing issues](https://github.com/MervinPraison/PraisonAIUI/issues) for duplicates',
'- Review [AGENTS.md](https://github.com/MervinPraison/PraisonAIUI/blob/main/AGENTS.md) for project conventions',
'',
'_A maintainer can trigger deeper analysis by commenting `@claude` on this issue._'
].join('\n')
});
}
if (isOwner) {
await github.rest.issues.addLabels({
issue_number: issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
labels: ['claude']
});
}
# ================================================================
# Job 2 — Full Claude Code implementation
# Triggers on: labeled `claude`, @claude comments, assignments
#
# Dedup: label + @claude comment on the same issue fires two events.
# Skip the comment trigger when `claude` was labelled within 3 minutes.
# ================================================================
claude-dedupe:
if: |
github.event_name == 'workflow_dispatch' ||
(
(github.event_name == 'issues' && github.event.action == 'opened' && contains(toJSON(github.event.issue.labels), 'claude')) ||
(github.event.action != 'opened' &&
(github.event.action != 'labeled' || github.event.label.name == 'claude') &&
(github.event_name != 'issue_comment' || contains(github.event.comment.body, '@claude')) &&
(
!contains(github.actor, '[bot]') ||
github.actor == 'github-actions[bot]' ||
github.actor == 'praisonai-triage-agent[bot]'
) &&
github.actor != 'dependabot[bot]' &&
github.actor != 'cursor[bot]' &&
github.actor != 'renovate[bot]'
)
)
runs-on: ubuntu-latest
outputs:
proceed: ${{ steps.check.outputs.proceed }}
permissions:
issues: read
pull-requests: read
steps:
- name: Skip duplicate comment after fresh claude label
id: check
uses: actions/github-script@v7
with:
script: |
if (context.eventName === 'workflow_dispatch') {
core.setOutput('proceed', 'true');
return;
}
if (context.eventName !== 'issue_comment' || context.payload.issue?.pull_request) {
core.setOutput('proceed', 'true');
return;
}
const issue = context.payload.issue;
const hasClaude = (issue.labels || []).some(l => l.name === 'claude');
if (!hasClaude) {
core.setOutput('proceed', 'true');
return;
}
const events = await github.rest.issues.listEvents({
owner: context.repo.owner,
repo: context.repo.repo,
issue_number: issue.number,
per_page: 30,
});
const windowMs = 3 * 60 * 1000;
const recentLabel = events.data.find(e =>
e.event === 'labeled' &&
e.label?.name === 'claude' &&
(Date.now() - new Date(e.created_at).getTime()) < windowMs
);
if (recentLabel) {
core.info('Skipping: claude label was added within 3 minutes — label trigger already handles this issue.');
core.setOutput('proceed', 'false');
return;
}
core.setOutput('proceed', 'true');
claude-response:
needs: claude-dedupe
env:
ISSUE_NUMBER_RESOLVED: ${{ github.event.issue.number || github.event.pull_request.number || inputs.issue_number }}
if: needs.claude-dedupe.outputs.proceed == 'true'
concurrency:
group: claude-${{ github.event.issue.number || github.event.pull_request.number || inputs.issue_number || github.run_id }}
cancel-in-progress: true
runs-on: ubuntu-latest
claude:
permissions:
contents: write
pull-requests: write
issues: write
actions: read
id-token: write
steps:
- name: Generate GitHub App Token
id: app-token
continue-on-error: true
uses: actions/create-github-app-token@v1
with:
app-id: ${{ secrets.CLAUDE_APP_ID }}
private-key: ${{ secrets.CLAUDE_APP_PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
- name: Resolve GitHub token
id: github-token
run: |
if [ -n "${{ steps.app-token.outputs.token }}" ]; then
echo "token=${{ steps.app-token.outputs.token }}" >> "$GITHUB_OUTPUT"
else
echo "token=${{ secrets.GITHUB_TOKEN }}" >> "$GITHUB_OUTPUT"
fi
- name: Check for Fork PR
id: check_fork
uses: actions/github-script@v7
with:
github-token: ${{ steps.github-token.outputs.token }}
script: |
const isPR = context.eventName === 'pull_request' ||
context.eventName === 'pull_request_review' ||
context.eventName === 'pull_request_review_comment' ||
(context.payload.issue && context.payload.issue.pull_request);
if (isPR) {
core.setOutput('is_pr_review', 'true');
const prNumber = context.payload.pull_request
? context.payload.pull_request.number
: context.payload.issue.number;
core.setOutput('pr_number', String(prNumber));
const pr = await github.rest.pulls.get({
owner: context.repo.owner,
repo: context.repo.repo,
pull_number: prNumber,
});
core.setOutput('pr_branch', pr.data.head.ref);
const headRepo = pr.data.head.repo?.full_name || context.repo.full_name;
core.setOutput('head_repo', headRepo);
let mergeState = '';
try {
const gql = await github.graphql(
`query($owner: String!, $repo: String!, $number: Int!) {
repository(owner: $owner, name: $repo) {
pullRequest(number: $number) { mergeStateStatus }
}
}`,
{ owner: context.repo.owner, repo: context.repo.repo, number: prNumber }
);
mergeState = (gql.repository.pullRequest.mergeStateStatus || '').toUpperCase();
} catch (e) {
core.warning(`GraphQL mergeStateStatus failed: ${e.message}`);
}
if (!mergeState) {
mergeState = (pr.data.mergeable_state || pr.data.merge_state_status || '').toUpperCase();
}
core.setOutput('merge_state', mergeState);
if (headRepo !== `${context.repo.owner}/${context.repo.repo}`) {
core.setOutput('is_fork', 'true');
core.setOutput('clone_url', pr.data.head.repo.clone_url);
core.setOutput('branch', pr.data.head.ref);
return;
}
core.setOutput('is_fork', 'false');
return;
}
core.setOutput('is_pr_review', 'false');
core.setOutput('is_fork', 'false');
core.setOutput('merge_state', '');
- name: Checkout repository
uses: actions/checkout@v4
with:
persist-credentials: false
fetch-depth: 0
token: ${{ steps.github-token.outputs.token }}
ref: ${{ steps.check_fork.outputs.is_pr_review == 'true' && format('refs/pull/{0}/head', steps.check_fork.outputs.pr_number) || github.ref }}
- name: Checkout PR branch for edits
if: steps.check_fork.outputs.is_pr_review == 'true' && steps.check_fork.outputs.is_fork != 'true'
env:
PR_BRANCH: ${{ steps.check_fork.outputs.pr_branch }}
run: |
git checkout "${PR_BRANCH}"
git config --local user.name "MervinPraison"
git config --local user.email "[email protected]"
echo "On branch: $(git branch --show-current)"
- name: Fetch PR branch and Setup Remote
if: steps.check_fork.outputs.is_pr_review == 'true' && steps.check_fork.outputs.is_fork == 'true'
env:
PR_BRANCH: ${{ steps.check_fork.outputs.pr_branch }}
run: |
git fetch origin pull/${{ steps.check_fork.outputs.pr_number }}/head:"${PR_BRANCH}"
git checkout "${PR_BRANCH}"
git remote set-url origin file://$(pwd)
- name: Run Claude PR final review
if: steps.check_fork.outputs.is_pr_review == 'true'
uses: anthropics/claude-code-action@beta
env:
GH_TOKEN: ${{ steps.github-token.outputs.token }}
with:
allowed_bots: 'praisonai-triage-agent[bot]'
claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
github_token: ${{ steps.github-token.outputs.token }}
model: claude-opus-4-8
trigger_phrase: "@claude"
timeout_minutes: "30"
direct_prompt: |
${{ steps.check_fork.outputs.is_fork == 'true' && 'CRITICAL: FORK PR — comment only, do not push.\n\n' || 'PR on MervinPraison/PraisonAIUI — SAME REPO, NOT A FORK. You HAVE push permission to this branch. NEVER say this is a fork.\n\n' }}
${{ steps.check_fork.outputs.is_fork != 'true' && steps.check_fork.outputs.merge_state == 'DIRTY' && format('PRIORITY — RESOLVE MERGE CONFLICTS (branch: {0}, PR #{1}):\n- DO NOT create a new branch or PR\n- git fetch origin main && git rebase origin/main\n- Resolve conflicts: keep this PR intent, merge in main newer logic\n- git add -A && git rebase --continue (repeat until done)\n- git push --force-with-lease origin {0}\n- Run targeted pytest/ruff on changed paths\n- Comment files resolved and what you skipped\n- Only after conflicts are resolved, proceed with review phases below if still needed\n\n', steps.check_fork.outputs.pr_branch, steps.check_fork.outputs.pr_number) || '' }}
${{ steps.check_fork.outputs.merge_state != 'DIRTY' && format('CONTEXT: PR #{0} on branch {1}. merge_state={2}.\n\n', steps.check_fork.outputs.pr_number, steps.check_fork.outputs.pr_branch, steps.check_fork.outputs.merge_state) || '' }}
You are the Lead Engineer doing FINAL review on this open PR in MervinPraison/PraisonAIUI.
Read ALL reviewer comments (CodeRabbit, Gemini, Qodo, Copilot, human). Follow AGENTS.md.
SCOPE: src/praisonaiui, src/frontend, tests, docs only.
**Phase 1 — Review** architecture, security, tests, backward compat, import-time budget.
Product value gate: change must strengthen PraisonAIUI UX (no scope creep). BLOCK params/modules/exports
duplicating existing UI capabilities or knobs with no live consumer. Core agent logic belongs in
praisonaiagents; agent-callable tools in PraisonAI-Tools — not here.
**Phase 2 — FIX (mandatory)** implement every VALID issue on THIS branch. Push commits before Phase 3.
Do NOT ask the author to fix. Do NOT open a new PR. Do NOT claim fork restrictions unless is_fork is true.
**Phase 3 — Verdict** approve only after fixes are pushed and tests pass; else request changes with specifics.
git config --global user.name "MervinPraison"
git config --global user.email "[email protected]"
gh auth setup-git
git status
git branch --show-current
allowed_tools: |
Bash(git:*)
Bash(python:*)
Bash(pip:*)
Bash(pytest:*)
Bash(ruff:*)
Bash(gh:*)
Bash(npm:*)
View
GlobTool
GrepTool
BatchTool
Edit
Replace
- name: Run Claude Code Action
if: steps.check_fork.outputs.is_pr_review != 'true'
uses: anthropics/claude-code-action@beta
env:
GH_TOKEN: ${{ steps.github-token.outputs.token }}
with:
allowed_bots: 'praisonai-triage-agent[bot]'
claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
github_token: ${{ steps.github-token.outputs.token }}
model: claude-opus-4-8
mode: ${{ github.event_name == 'workflow_dispatch' && 'agent' || 'tag' }}
trigger_phrase: "@claude"
label_trigger: "claude"
timeout_minutes: "30"
direct_prompt: |
${{ steps.check_fork.outputs.is_fork == 'true' && 'CRITICAL: THIS IS A PULL REQUEST FROM A FORK. YOU DO NOT HAVE PUSH PERMISSIONS TO THIS REPOSITORY. ONLY READ FILES, VALIDATE CODE, AND PROVIDE YOUR COMMENTS/FEEDBACK DIRECTLY IN THIS PR VIA COMMENTS. DO NOT ATTEMPT TO PUSH OR CREATE PULL REQUESTS.\n\n' || 'NOTE: The branch is under MervinPraison/PraisonAIUI (not a fork). You are able to make modifications and push directly to this branch.\n\n' }}
You are implementing changes in **MervinPraison/PraisonAIUI** (YAML-driven server-driven UI with agentic chat).
Follow `AGENTS.md` at the repository root. If a conflict arises between this prompt and `AGENTS.md`, `AGENTS.md` wins.
═══════════════════════════════════════════════════════════════
STEP 0 — GIT & GITHUB CLI
═══════════════════════════════════════════════════════════════
git config --global user.name "MervinPraison"
git config --global user.email "[email protected]"
gh auth setup-git
═══════════════════════════════════════════════════════════════
STEP 1 — READ GUIDELINES
═══════════════════════════════════════════════════════════════
Read `AGENTS.md` and any `docs/principles.md`. Key invariants:
• AIM: This package must stay LIGHTWEIGHT AND POWERFUL. Do a CRITICAL REVIEW at each stage
(issue triage, fix planning, PR review/implementation) and reject scope creep for the sake
of adding features. Prefer existing YAML config, plugins, feature modules, and server hooks
over new server.py/create_app() params or exports. Even when fixing a raised issue, keep the
fix minimal — do NOT expand surface area or add knobs with no live consumer.
• Cold import of `praisonaiui` must stay under 200 ms; module count must not grow.
• Heavy / optional deps (`mcp`, `langchain`, `llama_index`, `semantic_kernel`,
`slack_sdk`, `discord.py`, `botbuilder.core`, `openai`, `anthropic`, `mistralai`,
`google.generativeai`) must be **lazy-imported** — never eager.
• Public API is verb-first. New classes in `src/praisonaiui/message.py` ending
in `Message` with a verb prefix (Ask*, Do*, Request*, …) are forbidden —
use a module-level `async def` verb helper instead (see `prompt`, `error`).
• Deprecations must emit a runtime `warnings.warn(..., DeprecationWarning, stacklevel=2)`
on first access, not only in the docstring.
• Deterministic serialisation in anything reaching the wire (sorted keys, stable field
order). See existing `to_dict` patterns.
• Multi-agent + async safe: no global mutable state without a lock.
═══════════════════════════════════════════════════════════════
STEP 2 — ARCHITECTURE VALIDATION & ROUTING (MANDATORY before any code)
═══════════════════════════════════════════════════════════════
Answer these questions inline in your response, then proceed only if you pass:
ROUTING:
1. **Core aiui** (`src/praisonaiui/`) — protocols, message types, server, features.
2. **Features** (`src/praisonaiui/features/`) — opt-in capabilities (MCP, platforms, lifecycle, audio, auth, sharing, etc.).
3. **UI builders** (`src/praisonaiui/ui.py`) — declarative component helpers.
4. **Frontend** (`src/frontend/`) — React components / stores / SSE handlers.
5. **Tests** (`tests/unit/`, `tests/integration/`) — new features require both smoke and real-agentic tests.
6. **Docs** (`docs/`) — canonical name per feature, migration table for renames.
EXTERNAL REPOS (do NOT edit from within this repo — route the work to the right place and stop):
• Core SDK: `MervinPraison/PraisonAI` (`praisonaiagents`, `praisonai`) — agent runtime.
• Tools: `MervinPraison/PraisonAI-Tools` — tool implementations.
• Docs site: `MervinPraison/PraisonAIDocs` — product docs.
If the requested change belongs in an external repo, comment on the issue explaining
where it should live and DO NOT open a PR in this repo.
CONCEPTUAL CHECKS (if ANY fails — add a comment, do NOT open a PR):
• Does it duplicate an existing public symbol in `src/praisonaiui/__init__.py`?
• Does it duplicate existing UI capability via YAML config, plugins, or feature modules
instead of adding new server params? → Reject as scope creep; prefer existing patterns.
• Does it add core agent runtime logic here instead of praisonaiagents, or agent-callable
tools here instead of PraisonAI-Tools? → Route externally; do not implement here.
• Does it bloat server.py or create_app() with new params without a clear live consumer?
→ Prefer existing config/plugin hooks; only add params if absolutely required.
• Does it add new class-first message types (`class Ask*Message`, `class *Message`
with a verb in the name)? → Use a verb-first helper instead.
• Does it add a new hard dependency to `pyproject.toml` `[project] dependencies`?
→ Move to `[project.optional-dependencies]` and wrap imports lazily.
• Does it break backward compat (rename, signature change, removal)? → Add
a deprecation alias that emits `DeprecationWarning`; never hard-remove before the
documented removal version (e.g. `0.5.0`).
═══════════════════════════════════════════════════════════════
STEP 3 — IMPLEMENT
═══════════════════════════════════════════════════════════════
Single branch for the whole flow (reuse the same name in STEP 5):
BRANCH="claude/issue-${{ env.ISSUE_NUMBER_RESOLVED }}-$(date +%Y%m%d-%H%M%S)"
git checkout -b "$BRANCH"
Make minimal, focused changes:
• Stay within the files listed in the issue's "Scope" / "Files to change" section.
• Prefer extending existing modules over adding new ones.
• No emojis, no comments, no markdown except where the user asked for it.
• Python ≥ 3.10 style (`list[...]` not `List[...]`, `X | None` not `Optional[X]`).
═══════════════════════════════════════════════════════════════
STEP 4 — TEST / VERIFY
═══════════════════════════════════════════════════════════════
Run these targeted checks (NOT the full suite):
# 1. Lint on only the files you changed:
CHANGED=$(git diff --name-only origin/main...HEAD | grep -E '\.py$' | tr '\n' ' ')
[ -n "$CHANGED" ] && ruff check $CHANGED
# 2. Unit tests for the module(s) you touched:
pytest tests/unit/<your_new_test_file>.py -v --tb=short
# 3. Import-time invariant (MUST stay under 200 ms cold):
python -c "import time, sys; t=time.time(); import praisonaiui; dt=(time.time()-t)*1000; heavy=[m for m in sys.modules if any(h in m for h in ['langchain','llama_index','mcp','slack','discord','botbuilder','openai.','anthropic.','mistralai','google.generativeai'])]; print(f'{dt:.1f}ms modules={len(sys.modules)} leaked={heavy}'); assert heavy == [], f'Heavy deps leaked: {heavy}'"
If anything fails, fix the root cause (never skip, xfail, or delete the test).
═══════════════════════════════════════════════════════════════
STEP 5 — COMMIT, PUSH, OPEN PR (MANDATORY if you produced commits)
═══════════════════════════════════════════════════════════════
git add -A
git commit -m "feat: <short summary> (fixes #${{ env.ISSUE_NUMBER_RESOLVED }})"
git push -u origin "$BRANCH"
REPO="${{ github.repository }}"
COUNT=$(gh pr list --repo "$REPO" --head "$BRANCH" --json number --jq 'length')
if [ "$COUNT" -gt 0 ]; then
gh pr list --repo "$REPO" --head "$BRANCH" --json url --jq '.[0].url'
else
gh pr create --repo "$REPO" --base main --head "$BRANCH" \
--title "feat: <short title>" \
--body "Fixes #${{ env.ISSUE_NUMBER_RESOLVED }}
## Summary
<what changed in one paragraph>
## Acceptance criteria
<copy from the issue; tick boxes only with commit SHA + file path evidence>
## Test evidence
\`\`\`
<paste pytest output>
\`\`\`
"
fi
CRITICAL: Do NOT stop after `git push`. You MUST complete `gh pr create` (or print the
existing PR URL from the step above) and post that URL on the issue. Never ask a
human to open the PR manually.
═══════════════════════════════════════════════════════════════
STEP 6 — RESPOND TO REVIEW FEEDBACK (if this run was triggered by a review comment)
═══════════════════════════════════════════════════════════════
If you were invoked by a PR review comment, you are the Lead Engineer for this PR:
• Read ALL reviewer comments in chronological order (CodeRabbit → Copilot → human).
• Fix every VALID issue by pushing commits to THIS branch (do NOT open a new PR).
• Reply with a summary of which comments you addressed, which you disagreed with (and why).
• Never ship a fix that removes/weakens a test.
allowed_tools: |
Bash(git:*)
Bash(python:*)
Bash(pip:*)
Bash(pytest:*)
Bash(ruff:*)
Bash(gh:*)
Bash(npm:*)
Bash(npx:*)
Bash(python -m pytest:*)
Bash(python -m pip:*)
Bash(python -m ruff:*)
View
GlobTool
GrepTool
BatchTool
Edit
Replace
mcp__github__get_issue
mcp__github__get_issue_comments
mcp__github__update_issue
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
secrets: inherit
with:
hub_ref: v1.1.6
+2 -2
View File
@@ -25,9 +25,9 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: MervinPraison/github-automation-template/.github/actions/[email protected].1
- uses: MervinPraison/github-automation-template/.github/actions/[email protected].6
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
- name: Detect conflicting PRs and trigger Claude
uses: actions/github-script@v7
with:
+2 -2
View File
@@ -13,7 +13,7 @@ permissions:
jobs:
sync:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected].1
uses: MervinPraison/github-automation-template/.github/workflows/[email protected].6
secrets: inherit
with:
hub_ref: v1.1.1
hub_ref: v1.1.6
+1
View File
@@ -44,6 +44,7 @@ Thumbs.db
# Project specific
aiui/
*.log
*.err
.praison/
.gemini/
.internal/
+17 -2
View File
@@ -54,7 +54,8 @@ domain in dev — not localhost). Register in Recall:
- URL: `{PUBLIC_API_BASE_URL}/webhooks/recall`
- Events: `bot.joining_call`, `bot.in_call_recording`, `bot.done`, `bot.fatal`,
`recording.done`, `transcript.done`, `transcript.failed`
`recording.done`, `transcript.done`, `transcript.failed`, `transcript.data`
- Calendar (Part B): `calendar.sync_events`, `calendar.update` on the same URL
Use Recall MCP `create_webhook_endpoint` or the dashboard. Verify with
`send_test_webhook_endpoint`.
@@ -91,10 +92,24 @@ See `.env.example`:
Never commit real secrets. Reuse the existing REST key (`Mervin Praison`) when
its purpose is clear; otherwise create a purpose-named key via Recall MCP.
## Phase 2 Part B — live transcript + calendar auto-schedule
| Feature | API |
|---------|-----|
| Live transcript (SSE) | `GET /api/meetings/{id}/live-transcript` (`Accept: text/event-stream`) |
| Live transcript (poll) | Same URL, default JSON |
| Upcoming calendar events | `GET /api/recall/calendar/upcoming?hours=24` |
| Manual calendar sync | `POST /api/recall/calendar/sync` `{"calendar_id":"..."}` |
| Agent tool | `get_upcoming_meetings(hours=24)` in chat when Recall is configured |
Every new bot schedule **automatically** attaches `recording_config.realtime_endpoints`
pointing at `{PUBLIC_API_BASE_URL}/webhooks/recall` with `transcript.data` — no Recall
dashboard toggle for live transcript. Requires `PUBLIC_API_BASE_URL` in `.env`.
## Tests
From repo root:
```bash
pytest tests/unit/test_meeting_agent_example.py tests/unit/test_meeting_agent_pipeline.py tests/unit/test_meeting_agent_recall.py -v
pytest tests/unit/test_meeting_agent_example.py tests/unit/test_meeting_agent_pipeline.py tests/unit/test_meeting_agent_recall.py tests/unit/test_meeting_agent_part_b.py -v
```
+90 -22
View File
@@ -54,12 +54,17 @@ def _load_local_env() -> None:
_load_local_env()
aiui.set_custom_js(_EXAMPLE_DIR / "plugin.js")
from pipeline import retry_pipeline, run_ingest_pipeline
from recall_routes import (
api_meeting_live_transcript,
api_recall_bot_status,
api_recall_calendar_bootstrap,
api_recall_calendar_callback,
api_recall_calendar_status,
api_recall_calendar_sync,
api_recall_calendar_upcoming,
api_recall_cancel_bot,
api_recall_config_status,
api_recall_schedule_bot,
@@ -107,6 +112,19 @@ def _recall_schedule_tool():
return schedule_recall_bot
def _upcoming_meetings_tool():
from praisonai_tools.tools.decorator import tool
from meeting_integrations.recall.calendar_service import list_upcoming_calendar_meetings
@tool
def get_upcoming_meetings(hours: int = 24) -> list[dict]:
"""List upcoming calendar events with video links and bot schedule status."""
return list_upcoming_calendar_meetings(hours=hours)
return get_upcoming_meetings
def _load_meeting_tools():
"""Lazy-load meeting tools (keeps import-time deps light)."""
from praisonai_tools.tools.meeting_tools import (
@@ -132,6 +150,7 @@ def _load_meeting_tools():
]
if os.getenv("RECALL_API_KEY"):
tools.append(_recall_schedule_tool())
tools.append(_upcoming_meetings_tool())
return tools
@@ -275,6 +294,15 @@ def _meeting(meeting_id: str) -> dict | None:
return next((m for m in SAMPLE_MEETINGS if m["id"] == meeting_id), None)
def _pick_detail_meeting(meetings: list[dict[str, Any]]) -> dict[str, Any]:
"""Prefer an in-progress live meeting for the detail view."""
for live in ("live", "joining", "waiting_room"):
for m in meetings:
if (m.get("live_status") or "").lower() == live:
return m
return meetings[0] if meetings else SAMPLE_MEETINGS[0]
def _status_variant(status: str) -> str:
return {
"Ready": "default",
@@ -405,12 +433,13 @@ async def live_bot_page():
[
aiui.text(
"Send a Recall.ai Meeting Bot to a Zoom, Google Meet, or Teams URL. "
"When the call ends, the Phase 1 summarize + index pipeline runs "
"automatically on the Recall transcript."
"Real-time ``transcript.data`` chunks stream during the call; when the "
"call ends, the Phase 1 summarize + index pipeline runs on the full transcript."
),
aiui.alert(
"Requires RECALL_API_KEY, RECALL_WEBHOOK_VERIFICATION_SECRET, and "
"PUBLIC_API_BASE_URL (HTTPS tunnel) in the server environment.",
"PUBLIC_API_BASE_URL (HTTPS tunnel). Register ``transcript.data`` on the "
"same /webhooks/recall endpoint for live captions.",
variant="info",
title="Recall credentials",
),
@@ -422,8 +451,8 @@ async def live_bot_page():
language="bash",
),
aiui.text(
"Poll status: ``GET /api/recall/bots/{meeting_id}``. "
"Cancel: ``POST /api/recall/bots/{meeting_id}/cancel``."
"Live transcript SSE: ``GET /api/meetings/{meeting_id}/live-transcript`` "
"with ``Accept: text/event-stream``. Poll fallback: same URL without SSE."
),
]
)
@@ -431,29 +460,50 @@ async def live_bot_page():
@aiui.page("calendar", title="Calendar", icon="📅", group="Meetings", order=5)
async def calendar_page():
upcoming: list[dict[str, Any]] = []
status_note = "Set RECALL_API_KEY to load calendar status."
if os.getenv("RECALL_API_KEY"):
try:
from meeting_integrations.recall.calendar_service import list_upcoming_calendar_meetings
upcoming = list_upcoming_calendar_meetings(hours=48)
status_note = f"{len(upcoming)} upcoming event(s) with video links (48h window)."
except Exception as exc: # noqa: BLE001
status_note = f"Calendar API unavailable: {exc}"
rows = [
[
e.get("title", "—"),
str(e.get("start", ""))[:16],
e.get("platform") or "—",
"Yes" if e.get("bot_scheduled") else "No",
]
for e in upcoming[:10]
]
return aiui.layout(
[
aiui.text(
"Google Calendar V2 auto-schedules Recall bots for eligible future "
"events with video links (default: CALENDAR_AUTO_RECORD=true)."
"events with video links when CALENDAR_AUTO_RECORD=true (default)."
),
aiui.alert(status_note, variant="info", title="Calendar status"),
aiui.table(
headers=["Title", "Start", "Platform", "Bot scheduled"],
rows=rows or [["—", "—", "—", "—"]],
),
aiui.alert(
"Complete Calendar V2 setup via Recall MCP "
"(start_calendar_integration_setup) using "
"``PUBLIC_API_BASE_URL/api/recall/calendar/callback`` as "
"production_redirect_uri, then bootstrap with "
"``POST /api/recall/calendar/setup/bootstrap``.",
"Complete Calendar V2 setup via Recall MCP, bootstrap OAuth callback, "
"then register calendar webhooks (calendar.sync_events, calendar.update) "
"on /webhooks/recall.",
variant="info",
title="Calendar V2 setup",
title="Setup",
),
aiui.code_block(
"GET /api/recall/calendar/status",
"GET /api/recall/calendar/status\n"
"GET /api/recall/calendar/upcoming?hours=24\n"
"POST /api/recall/calendar/sync {\"calendar_id\":\"...\"}",
language="bash",
),
aiui.text(
"Recording opt-in: connecting a calendar syncs events; bots are "
"sent only when CALENDAR_AUTO_RECORD is enabled (default on)."
),
]
)
@@ -466,7 +516,8 @@ async def meetings_page():
m["title"],
m["date"],
m["status"],
str(len(m["actions"])),
m.get("live_status") or "—",
m["id"],
m["duration"],
]
for m in meetings
@@ -475,7 +526,7 @@ async def meetings_page():
[
aiui.metric("Meetings", value=len(meetings)),
aiui.table(
headers=["Title", "Date", "Status", "Actions", "Duration"],
headers=["Title", "Date", "Status", "Live", "ID", "Duration"],
rows=rows,
),
]
@@ -485,18 +536,28 @@ async def meetings_page():
@aiui.page("meeting-detail", title="Meeting detail", icon="📝", group="Meetings", order=4)
async def meeting_detail_page():
meetings = _list_meeting_views()
meeting = meetings[0]
meeting = _pick_detail_meeting(meetings)
action_rows = [[a["task"], a["owner"], a["due"]] for a in meeting["actions"]]
live = meeting.get("live_status") or ""
live = (meeting.get("live_status") or "").lower()
header: list[Any] = [
aiui.text(meeting["title"]),
aiui.badge(meeting["status"], variant=_status_variant(meeting["status"])),
aiui.text(f"ID: {meeting['id']}"),
]
if live:
header.append(aiui.badge(live.title(), variant="outline"))
variant = "default" if live == "live" else "outline"
header.append(aiui.badge(live.title(), variant=variant))
body: list[Any] = []
if meeting.get("error"):
body.append(aiui.alert(meeting["error"], variant="warning"))
if live == "live":
body.append(
aiui.alert(
"Live transcript auto-refreshes every 3 seconds on this page.",
variant="info",
title="Live",
)
)
return aiui.layout(
header
+ body
@@ -646,6 +707,13 @@ _meeting_routes = [
methods=["POST"],
),
Route("/api/recall/calendar/status", api_recall_calendar_status, methods=["GET"]),
Route("/api/recall/calendar/upcoming", api_recall_calendar_upcoming, methods=["GET"]),
Route("/api/recall/calendar/sync", api_recall_calendar_sync, methods=["POST"]),
Route(
"/api/meetings/{meeting_id}/live-transcript",
api_meeting_live_transcript,
methods=["GET"],
),
]
# Prepend so custom /api/* routes win over the SPA catch-all from create_app().
app.routes[0:0] = _meeting_routes
@@ -0,0 +1,261 @@
"""Google Calendar V2 sync and auto-schedule via Recall Calendar API."""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from typing import Any
from meeting_integrations.recall.client import RecallAPIError, RecallClient
from meeting_integrations.recall.config import RecallSettings
from meeting_integrations.recall.store import RecallStore
logger = logging.getLogger(__name__)
def _parse_iso(value: str | None) -> datetime | None:
if not value:
return None
try:
normalized = value.replace("Z", "+00:00")
if normalized.endswith("+0000"):
normalized = normalized[:-5] + "+00:00"
return datetime.fromisoformat(normalized)
except ValueError:
return None
def _event_title(event: dict[str, Any]) -> str:
raw = event.get("raw") or {}
if isinstance(raw, dict):
summary = raw.get("summary") or raw.get("subject")
if summary:
return str(summary)
return "Calendar meeting"
def _deduplication_key(event: dict[str, Any]) -> str:
start = str(event.get("start_time") or "")
url = str(event.get("meeting_url") or "")
return f"{start}-{url}"
def is_eligible_for_auto_record(event: dict[str, Any]) -> bool:
"""Return True when Recall should schedule a bot for this calendar event."""
if event.get("is_deleted"):
return False
if not event.get("meeting_url"):
return False
end = _parse_iso(str(event.get("end_time") or ""))
if end and end < datetime.now(timezone.utc):
return False
raw = event.get("raw") or {}
if isinstance(raw, dict):
status = str(raw.get("status") or "").lower()
if status == "cancelled":
return False
return True
def list_upcoming_calendar_meetings(
*,
hours: int = 24,
settings: RecallSettings | None = None,
client: RecallClient | None = None,
) -> list[dict[str, Any]]:
"""List upcoming Recall calendar events with video links (for UI + agent tool)."""
from meeting_integrations.recall.config import load_recall_settings
settings = settings or load_recall_settings()
client = client or RecallClient(settings)
try:
events = client.list_calendar_events(is_deleted=False)
except RecallAPIError as exc:
logger.warning("Calendar events unavailable: %s", exc)
return []
now = datetime.now(timezone.utc)
horizon = now + timedelta(hours=max(1, hours))
rows: list[dict[str, Any]] = []
for event in events:
if not is_eligible_for_auto_record(event):
continue
start = _parse_iso(str(event.get("start_time") or ""))
if start and start > horizon:
continue
bots = event.get("bots") or []
rows.append(
{
"event_id": event.get("id"),
"title": _event_title(event),
"start": event.get("start_time"),
"end": event.get("end_time"),
"platform": (event.get("meeting_platform") or {}).get("name")
if isinstance(event.get("meeting_platform"), dict)
else event.get("meeting_platform"),
"meeting_url": event.get("meeting_url"),
"bot_scheduled": bool(bots),
"calendar_id": event.get("calendar_id"),
}
)
rows.sort(key=lambda r: str(r.get("start") or ""))
return rows
def _ensure_local_meeting_for_event(
event: dict[str, Any],
*,
settings: RecallSettings,
) -> str:
"""Create or reuse a local meeting row for a calendar event."""
event_id = str(event.get("id") or "")
if RecallStore.is_event_scheduled(event_id):
row = RecallStore.get_scheduled_bot(event_id)
if row and row.get("meeting_id"):
return str(row["meeting_id"])
from praisonai_tools.tools.meeting_tools import save_meeting
title = _event_title(event)
url = str(event.get("meeting_url") or "")
saved = save_meeting.__wrapped__(
title=title,
source=url,
metadata={
"status": "scheduled",
"live_status": "scheduled",
"meeting_url": url,
"source": "calendar",
"recall_calendar_event_id": event_id,
},
)
if "error" in saved:
raise RecallAPIError(0, saved["error"])
return str(saved["meeting_id"])
def sync_calendar_events(
calendar_id: str,
*,
updated_since: str | None = None,
settings: RecallSettings,
client: RecallClient | None = None,
auto_record: bool | None = None,
) -> dict[str, Any]:
"""Fetch changed calendar events and schedule/remove bots per policy."""
client = client or RecallClient(settings)
should_record = (
settings.calendar_auto_record if auto_record is None else auto_record
)
events = client.list_calendar_events(
calendar_id=calendar_id,
updated_at_gte=updated_since,
)
scheduled = 0
removed = 0
skipped = 0
errors: list[str] = []
for event in events:
event_id = str(event.get("id") or "")
if not event_id:
continue
eligible = is_eligible_for_auto_record(event) and should_record
has_bots = bool(event.get("bots"))
if not eligible:
if has_bots and (event.get("is_deleted") or not should_record):
try:
client.remove_bot_from_calendar_event(event_id)
removed += 1
except RecallAPIError as exc:
errors.append(f"{event_id}: remove failed: {exc.detail}")
else:
skipped += 1
continue
if RecallStore.is_event_scheduled(event_id) and has_bots:
skipped += 1
continue
try:
meeting_id = _ensure_local_meeting_for_event(event, settings=settings)
bot_config = client.build_bot_config(
meeting_id=meeting_id,
title=_event_title(event),
)
updated = client.schedule_bot_for_calendar_event(
event_id,
deduplication_key=_deduplication_key(event),
bot_config=bot_config,
)
bot_id = None
bots = updated.get("bots") or []
if bots and isinstance(bots[0], dict):
bot_id = bots[0].get("bot_id") or bots[0].get("id")
RecallStore.record_scheduled_bot(event_id, meeting_id, bot_id)
from pipeline import merge_meeting_metadata
merge_meeting_metadata(
meeting_id,
{
"recall_calendar_event_id": event_id,
"recall_bot_id": bot_id,
"status": "scheduled",
"live_status": "scheduled",
},
)
scheduled += 1
except RecallAPIError as exc:
logger.warning("Calendar schedule failed for %s: %s", event_id, exc.detail)
errors.append(f"{event_id}: {exc.detail}")
return {
"calendar_id": calendar_id,
"fetched": len(events),
"scheduled": scheduled,
"removed": removed,
"skipped": skipped,
"errors": errors,
}
def process_calendar_webhook(
event_type: str,
payload: dict[str, Any],
*,
settings: RecallSettings,
client: RecallClient | None = None,
) -> None:
"""Handle Recall Calendar V2 webhooks (calendar.update, calendar.sync_events)."""
data = payload.get("data") or {}
if not isinstance(data, dict):
return
calendar_id = str(data.get("calendar_id") or "")
if not calendar_id:
logger.warning("Calendar webhook %s missing calendar_id", event_type)
return
if event_type == "calendar.update":
logger.info("Calendar %s updated — refresh status in UI", calendar_id)
return
if event_type == "calendar.sync_events":
updated_since = str(data.get("last_updated_ts") or "") or None
result = sync_calendar_events(
calendar_id,
updated_since=updated_since,
settings=settings,
client=client,
)
logger.info("Calendar sync %s: %s", calendar_id, result)
return
logger.debug("Ignoring calendar webhook event %s", event_type)
@@ -91,6 +91,7 @@ class RecallClient:
meeting_id: str,
title: str,
join_at: str | None = None,
enable_realtime_transcription: bool = True,
) -> dict[str, Any]:
"""Schedule a Recall bot; attach local meeting_id in metadata for webhook routing."""
body: dict[str, Any] = {
@@ -100,12 +101,44 @@ class RecallClient:
}
if join_at:
body["join_at"] = join_at
if enable_realtime_transcription:
recording_config = self._settings.realtime_recording_config()
if recording_config:
body["recording_config"] = recording_config
logger.info(
"Live transcript enabled for meeting_id=%s → %s",
meeting_id,
self._settings.webhook_url,
)
else:
logger.warning(
"Live transcript skipped for meeting_id=%s — set PUBLIC_API_BASE_URL",
meeting_id,
)
logger.info(
"Creating Recall bot for meeting_id=%s (url redacted)",
meeting_id,
)
return self._request("POST", "/api/v1/bot/", json_body=body)
def build_bot_config(
self,
*,
meeting_id: str,
title: str,
enable_realtime_transcription: bool = True,
) -> dict[str, Any]:
"""Bot config payload for Calendar V2 schedule-bot requests."""
config: dict[str, Any] = {
"bot_name": self._settings.bot_name,
"metadata": {"meeting_id": meeting_id, "title": title},
}
if enable_realtime_transcription:
recording_config = self._settings.realtime_recording_config()
if recording_config:
config["recording_config"] = recording_config
return config
def get_bot(self, bot_id: str) -> dict[str, Any]:
return self._request("GET", f"/api/v1/bot/{bot_id}/")
@@ -178,3 +211,80 @@ class RecallClient:
response = client.get(url)
response.raise_for_status()
return response.json()
def list_calendars(self, *, status: str | None = "connected") -> list[dict[str, Any]]:
params: dict[str, str] = {}
if status:
params["status"] = status
results: list[dict[str, Any]] = []
cursor: str | None = None
while True:
query = dict(params)
if cursor:
query["cursor"] = cursor
path = "/api/v2/calendars/"
if query:
from urllib.parse import urlencode
path = f"{path}?{urlencode(query)}"
page = self._request("GET", path)
batch = page.get("results") or []
if isinstance(batch, list):
results.extend(item for item in batch if isinstance(item, dict))
cursor = page.get("next")
if not cursor:
break
return results
def list_calendar_events(
self,
*,
calendar_id: str | None = None,
updated_at_gte: str | None = None,
is_deleted: bool | None = False,
) -> list[dict[str, Any]]:
from urllib.parse import urlencode
params: dict[str, str] = {}
if calendar_id:
params["calendar_id"] = calendar_id
if updated_at_gte:
params["updated_at__gte"] = updated_at_gte
if is_deleted is not None:
params["is_deleted"] = "true" if is_deleted else "false"
results: list[dict[str, Any]] = []
cursor: str | None = None
while True:
query = dict(params)
if cursor:
query["cursor"] = cursor
path = f"/api/v2/calendar-events/?{urlencode(query)}" if query else "/api/v2/calendar-events/"
page = self._request("GET", path)
batch = page.get("results") or []
if isinstance(batch, list):
results.extend(item for item in batch if isinstance(item, dict))
cursor = page.get("next")
if not cursor:
break
return results
def schedule_bot_for_calendar_event(
self,
event_id: str,
*,
deduplication_key: str,
bot_config: dict[str, Any],
) -> dict[str, Any]:
body = {
"deduplication_key": deduplication_key,
"bot_config": bot_config,
}
return self._request(
"POST",
f"/api/v2/calendar-events/{event_id}/bot/",
json_body=body,
)
def remove_bot_from_calendar_event(self, event_id: str) -> dict[str, Any]:
return self._request("DELETE", f"/api/v2/calendar-events/{event_id}/bot/")
@@ -4,6 +4,7 @@ from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse
@@ -35,6 +36,31 @@ class RecallSettings:
def calendar_callback_url(self) -> str:
return f"{self.public_api_base_url.rstrip('/')}/api/recall/calendar/callback"
def realtime_recording_config(self) -> dict[str, Any] | None:
"""Recording config with live ``transcript.data`` webhooks (no dashboard toggle)."""
if not self.public_api_base_url:
return None
return {
"transcript": {
"provider": {
"recallai_streaming": {
"mode": "prioritize_low_latency",
"language_code": "en",
}
},
"diarization": {
"use_separate_streams_when_available": True,
},
},
"realtime_endpoints": [
{
"type": "webhook",
"url": self.webhook_url,
"events": ["transcript.data", "transcript.partial_data"],
}
],
}
def _require(name: str) -> str:
value = os.getenv(name, "").strip()
@@ -0,0 +1,83 @@
"""In-process live transcript fan-out for SSE subscribers."""
from __future__ import annotations
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Any
@dataclass
class LiveTranscriptState:
"""Latest live transcript snapshot for a meeting."""
lines: list[str] = field(default_factory=list)
text: str = ""
live_status: str = ""
_lock = threading.Lock()
_states: dict[str, LiveTranscriptState] = {}
_subscribers: dict[str, list[threading.Event]] = defaultdict(list)
def get_live_state(meeting_id: str) -> LiveTranscriptState | None:
with _lock:
return _states.get(meeting_id)
def publish_live_chunk(
meeting_id: str,
*,
line: str,
full_text: str,
live_status: str = "live",
) -> None:
"""Record a new utterance and notify SSE waiters."""
with _lock:
state = _states.setdefault(meeting_id, LiveTranscriptState())
if line:
state.lines.append(line)
state.text = full_text
state.live_status = live_status
waiters = list(_subscribers.get(meeting_id, []))
for event in waiters:
event.set()
def set_live_status(meeting_id: str, live_status: str) -> None:
with _lock:
state = _states.setdefault(meeting_id, LiveTranscriptState())
state.live_status = live_status
waiters = list(_subscribers.get(meeting_id, []))
for event in waiters:
event.set()
def wait_for_update(meeting_id: str, timeout: float = 25.0) -> LiveTranscriptState | None:
"""Block until the meeting transcript changes or timeout (for SSE long-poll)."""
event = threading.Event()
with _lock:
_subscribers[meeting_id].append(event)
state = _states.get(meeting_id)
event.wait(timeout=timeout)
with _lock:
subs = _subscribers.get(meeting_id, [])
if event in subs:
subs.remove(event)
return _states.get(meeting_id)
def snapshot_for_api(meeting_id: str, record: dict[str, Any] | None = None) -> dict[str, Any]:
"""Merge in-memory live state with persisted meeting metadata."""
meta = (record or {}).get("metadata") or {}
live = get_live_state(meeting_id)
text = (live.text if live and live.text else meta.get("transcript")) or ""
return {
"meeting_id": meeting_id,
"live_status": (live.live_status if live and live.live_status else meta.get("live_status"))
or "",
"transcript": text,
"line_count": len([ln for ln in text.splitlines() if ln.strip()]),
}
@@ -8,11 +8,57 @@ from typing import Any
from meeting_integrations.recall.client import RecallAPIError, RecallClient
from meeting_integrations.recall.config import RecallSettings
from meeting_integrations.recall.store import RecallStore, webhook_event_key
from meeting_integrations.recall.transcript import transcript_download_to_text
from meeting_integrations.recall.transcript import transcript_data_to_line, transcript_download_to_text
logger = logging.getLogger(__name__)
def _speaker_prefix(line: str) -> str:
if ":" in line:
return line.split(":", 1)[0] + ":"
return ""
def _upsert_live_transcript(meeting_id: str, line: str, *, partial: bool = False) -> str:
"""Merge one live caption line; partial updates replace the last same-speaker line."""
from praisonai_tools.tools.meeting_tools import get_meeting
from meeting_integrations.recall.live import publish_live_chunk
from pipeline import _metadata_lock
# Hold the merge lock across the read too: streaming transcript events are
# dispatched one-thread-each, so an unsynchronized read-append-write would
# let concurrent captions clobber each other and drop lines.
with _metadata_lock:
record = get_meeting.__wrapped__(meeting_id)
meta = record.get("metadata") or {} if isinstance(record, dict) else {}
existing = str(meta.get("transcript") or "")
lines = existing.splitlines()
prefix = _speaker_prefix(line)
if lines and lines[-1] == line:
full = existing
elif partial and lines and prefix and lines[-1].startswith(prefix):
lines[-1] = line
full = "\n".join(lines)
elif not partial and lines and prefix and lines[-1].startswith(prefix):
# Final utterance replaces the in-progress partial for this speaker.
lines[-1] = line
full = "\n".join(lines)
else:
full = f"{existing}\n{line}".strip() if existing else line
_merge_metadata(
meeting_id,
{"transcript": full, "live_status": "live", "status": "live"},
)
publish_live_chunk(meeting_id, line=line, full_text=full, live_status="live")
return full
def _append_live_transcript(meeting_id: str, line: str) -> str:
return _upsert_live_transcript(meeting_id, line, partial=False)
def _merge_metadata(meeting_id: str, patch: dict[str, Any]) -> None:
from pipeline import merge_meeting_metadata
@@ -73,11 +119,29 @@ def process_recall_webhook(
return
if event_type == "bot.in_call_recording" and meeting_id:
_merge_metadata(meeting_id, {"live_status": "live", "status": "live"})
_merge_metadata(meeting_id, {"live_status": "live", "status": "live", "error": ""})
from meeting_integrations.recall.live import set_live_status
set_live_status(meeting_id, "live")
return
if event_type == "transcript.data" and meeting_id:
line = transcript_data_to_line(payload)
if line:
_append_live_transcript(meeting_id, line)
return
if event_type == "transcript.partial_data" and meeting_id:
line = transcript_data_to_line(payload)
if line:
_upsert_live_transcript(meeting_id, line, partial=True)
return
if event_type == "bot.call_ended" and meeting_id:
_merge_metadata(meeting_id, {"live_status": "ended"})
from meeting_integrations.recall.live import set_live_status
set_live_status(meeting_id, "ended")
return
if event_type == "bot.fatal" and meeting_id:
@@ -88,6 +88,7 @@ def schedule_recall_bot(
},
)
realtime = settings.realtime_recording_config() is not None
logger.info("Scheduled Recall bot %s for meeting %s", bot_id, meeting_id)
return {
"meeting_id": meeting_id,
@@ -96,6 +97,8 @@ def schedule_recall_bot(
"live_status": "scheduled",
"title": meeting_title,
"meeting_url": url,
"realtime_transcript_enabled": realtime,
"realtime_webhook_url": settings.webhook_url if realtime else None,
}
@@ -9,6 +9,8 @@ from contextlib import closing
from pathlib import Path
from typing import Any
from meeting_integrations.recall.client import RecallClient
def _state_db_path() -> str:
env = os.getenv("PRAISONAI_MEETINGS_DIR")
@@ -54,6 +56,32 @@ def _connect() -> sqlite3.Connection:
def webhook_event_key(event_type: str, payload: dict[str, Any]) -> str:
"""Build a stable idempotency key from webhook metadata."""
data = payload.get("data") or {}
# Streaming transcript events fire many times per bot — never dedupe on bot_id alone.
if event_type in ("transcript.data", "transcript.partial_data"):
inner = data.get("data") if isinstance(data, dict) else {}
bot_id = RecallClient.extract_bot_id(payload) or ""
if isinstance(inner, dict):
words = inner.get("words")
if isinstance(words, list) and words:
first = words[0] if isinstance(words[0], dict) else {}
last = words[-1] if isinstance(words[-1], dict) else {}
word_text = " ".join(
w.get("text", "") for w in words if isinstance(w, dict)
).strip()
return (
f"{event_type}:{bot_id}:"
f"{first.get('start', first.get('start_time', ''))}:"
f"{last.get('end', last.get('end_time', ''))}:"
f"{word_text}"
)
text = str(inner.get("text") or "").strip()
start = inner.get("start") or inner.get("start_time") or ""
end = inner.get("end") or inner.get("end_time") or ""
speaker = inner.get("speaker") or inner.get("speaker_name") or ""
return f"{event_type}:{bot_id}:{start}:{end}:{speaker}:{text}"
return f"{event_type}:{json.dumps(payload, sort_keys=True)}"
for part in (
payload.get("id"),
data.get("id") if isinstance(data, dict) else None,
@@ -164,3 +192,22 @@ class RecallStore:
(event_id,),
).fetchone()
return row is not None
@staticmethod
def get_scheduled_bot(event_id: str) -> dict[str, Any] | None:
with closing(_connect()) as conn:
row = conn.execute(
"""
SELECT event_id, meeting_id, bot_id, scheduled_at
FROM recall_scheduled_bots WHERE event_id = ?
""",
(event_id,),
).fetchone()
if not row:
return None
return {
"event_id": row[0],
"meeting_id": row[1],
"bot_id": row[2],
"scheduled_at": row[3],
}
@@ -5,6 +5,34 @@ from __future__ import annotations
from typing import Any
def transcript_data_to_line(payload: dict[str, Any]) -> str:
"""Flatten a verified ``transcript.data`` webhook payload to one line."""
data = payload.get("data") or {}
if not isinstance(data, dict):
return ""
inner = data.get("data") or data
if not isinstance(inner, dict):
return ""
words = inner.get("words")
if isinstance(words, list) and words:
text = " ".join(
w.get("text", "") for w in words if isinstance(w, dict)
).strip()
else:
text = str(inner.get("text") or "").strip()
if not text:
return ""
speaker = inner.get("speaker") or inner.get("speaker_name")
participant = inner.get("participant")
if not speaker and isinstance(participant, dict):
speaker = participant.get("name")
return f"{speaker}: {text}" if speaker else text
def transcript_download_to_text(data: Any) -> str:
"""Flatten Recall transcript JSON into a readable multi-line string."""
if isinstance(data, str):
+21 -13
View File
@@ -10,12 +10,19 @@ from __future__ import annotations
import json
import logging
import re
import threading
from contextlib import closing
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# Serializes metadata read-modify-write. Webhooks are dispatched one-thread-each
# (asyncio.to_thread), so concurrent transcript events would otherwise clobber
# each other's appends. Reentrant so callers can hold it across their own
# read + merge (see processor._upsert_live_transcript).
_metadata_lock = threading.RLock()
AUDIO_SUFFIXES = {".mp3", ".mp4", ".m4a", ".wav", ".webm", ".ogg", ".mpeg", ".mpga"}
_MEETING_ID_RE = re.compile(
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
@@ -27,19 +34,20 @@ def merge_meeting_metadata(meeting_id: str, patch: dict[str, Any]) -> dict[str,
"""Merge keys into a meeting's metadata blob."""
from praisonai_tools.tools import meeting_tools as mt
record = mt.get_meeting.__wrapped__(meeting_id)
if "error" in record:
return record
meta = {**(record.get("metadata") or {}), **patch}
try:
with closing(mt._connect()) as conn, conn:
conn.execute(
"UPDATE meetings SET metadata = ? WHERE meeting_id = ?",
(json.dumps(meta, sort_keys=True), meeting_id),
)
except Exception as exc: # noqa: BLE001
logger.error("merge_meeting_metadata failed: %s", exc)
return {"error": str(exc)}
with _metadata_lock:
record = mt.get_meeting.__wrapped__(meeting_id)
if "error" in record:
return record
meta = {**(record.get("metadata") or {}), **patch}
try:
with closing(mt._connect()) as conn, conn:
conn.execute(
"UPDATE meetings SET metadata = ? WHERE meeting_id = ?",
(json.dumps(meta, sort_keys=True), meeting_id),
)
except Exception as exc: # noqa: BLE001
logger.error("merge_meeting_metadata failed: %s", exc)
return {"error": str(exc)}
return {"meeting_id": meeting_id, "metadata": meta}
+57
View File
@@ -0,0 +1,57 @@
/**
* Meeting Agent dashboard hooks — auto-refresh live transcript on meeting-detail.
*/
(function () {
'use strict';
let pollTimer = null;
function stopPolling() {
if (pollTimer) {
clearInterval(pollTimer);
pollTimer = null;
}
}
async function refreshLiveTranscript() {
const host = document.querySelector('[data-page="meeting-detail"]');
if (!host) return;
const code = host.querySelector('.db-code-block code');
if (!code) return;
try {
const meetingsRes = await fetch('/api/meetings');
if (!meetingsRes.ok) return;
const meetingsJson = await meetingsRes.json();
const meetings = meetingsJson.meetings || [];
const live = meetings.find((m) =>
['live', 'joining', 'waiting_room'].includes(String(m.live_status || '').toLowerCase())
);
if (!live || !live.id) return;
const snapRes = await fetch(`/api/meetings/${live.id}/live-transcript`);
if (!snapRes.ok) return;
const snap = await snapRes.json();
const text = snap.transcript || '(empty)';
if (code.textContent !== text) {
code.textContent = text;
}
} catch (_err) {
/* ignore transient network errors */
}
}
function startPolling() {
stopPolling();
refreshLiveTranscript();
pollTimer = setInterval(refreshLiveTranscript, 3000);
}
window.addEventListener('aiui:page-change', (event) => {
stopPolling();
if (event.detail && event.detail.pageId === 'meeting-detail') {
startPolling();
}
});
})();
@@ -0,0 +1,61 @@
# Poll live transcript for a meeting (Part B dev test).
# Usage:
# .\poll_live_transcript.ps1
# .\poll_live_transcript.ps1 -MeetingId "aecf34bc-7194-4123-9c67-a298b56011da"
# .\poll_live_transcript.ps1 -IntervalSec 2
param(
[string]$MeetingId = "",
[int]$Port = 8000,
[int]$IntervalSec = 3
)
$ErrorActionPreference = "Stop"
$base = "http://127.0.0.1:$Port"
if (-not $MeetingId) {
$meetings = Invoke-RestMethod -Uri "$base/api/meetings"
$live = $meetings.meetings | Where-Object { @("live","joining","waiting_room") -contains ($_.live_status).ToLower() } | Select-Object -First 1
if ($live) {
$MeetingId = $live.id
} elseif ($meetings.meetings.Count -gt 0) {
$MeetingId = $meetings.meetings[0].id
} else {
throw "No meetings found. Schedule a bot first."
}
}
$url = "$base/api/meetings/$MeetingId/live-transcript"
Write-Host "Polling $url every ${IntervalSec}s (Ctrl+C to stop)"
Write-Host ""
$lastCount = -1
$lastText = ""
while ($true) {
try {
$r = Invoke-RestMethod -Uri $url -TimeoutSec 10
$count = [int]$r.line_count
$status = $r.live_status
$text = [string]$r.transcript
if ($count -ne $lastCount -or $text -ne $lastText) {
$ts = Get-Date -Format "HH:mm:ss"
Write-Host "[$ts] live_status=$status lines=$count"
if ($text) {
Write-Host "---"
Write-Host $text
Write-Host "---"
} else {
Write-Host "(empty transcript)"
}
Write-Host ""
$lastCount = $count
$lastText = $text
}
} catch {
Write-Host "[$(Get-Date -Format 'HH:mm:ss')] ERROR: $($_.Exception.Message)"
}
Start-Sleep -Seconds $IntervalSec
}
+123 -18
View File
@@ -9,10 +9,16 @@ import os
from typing import Any
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse, Response
from starlette.responses import JSONResponse, PlainTextResponse, Response, StreamingResponse
from meeting_integrations.recall.calendar import forward_calendar_callback
from meeting_integrations.recall.calendar_service import (
list_upcoming_calendar_meetings,
process_calendar_webhook,
sync_calendar_events,
)
from meeting_integrations.recall.config import RecallConfigError, load_recall_settings
from meeting_integrations.recall.live import snapshot_for_api, wait_for_update
from meeting_integrations.recall.processor import process_recall_webhook
from meeting_integrations.recall.service import cancel_recall_bot, schedule_recall_bot
from meeting_integrations.recall.store import RecallStore
@@ -22,9 +28,16 @@ logger = logging.getLogger(__name__)
_calendar_regional_uri: str | None = None
# Keep strong references to fire-and-forget webhook tasks so the event loop does
# not garbage-collect them mid-execution (asyncio only holds weak references).
_background_tasks: set[asyncio.Task[Any]] = set()
# asyncio keeps only weak references to tasks, so a bare create_task() can be
# garbage-collected mid-flight and silently drop a webhook event. Hold a strong
# reference until each background task finishes.
_background_tasks: set[asyncio.Task] = set()
def _dispatch_background(coro) -> None:
task = asyncio.create_task(coro)
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
def set_calendar_regional_callback_uri(uri: str | None) -> None:
@@ -107,11 +120,20 @@ async def webhook_recall(request: Request) -> Response:
return JSONResponse({"error": "invalid JSON"}, status_code=400)
event_type = str(payload.get("event") or payload.get("type") or "unknown")
task = asyncio.create_task(
if event_type.startswith("calendar."):
_dispatch_background(
asyncio.to_thread(
process_calendar_webhook,
event_type,
payload,
settings=settings,
)
)
return PlainTextResponse("ok", status_code=200)
_dispatch_background(
asyncio.to_thread(process_recall_webhook, event_type, payload, settings=settings)
)
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
return PlainTextResponse("ok", status_code=200)
@@ -184,30 +206,113 @@ async def api_recall_calendar_status(_request: Request) -> JSONResponse:
"off",
}
public_url = os.getenv("PUBLIC_API_BASE_URL", "").strip()
calendars: list[dict[str, Any]] = []
try:
from meeting_integrations.recall.client import RecallClient
settings = load_recall_settings()
calendars = RecallClient(settings).list_calendars(status="connected")
except Exception as exc: # noqa: BLE001
logger.debug("Could not list calendars: %s", exc)
return JSONResponse(
{
"calendar_v2": {
"auto_record_eligible_events": auto_record,
"opt_in_rule": (
"When CALENDAR_AUTO_RECORD=true (default), Recall Calendar V2 "
"schedules bots for every eligible future calendar event with a "
"video link. Set CALENDAR_AUTO_RECORD=false to disable auto-schedule."
),
"connected_calendars": len(calendars),
"calendars": [
{
"id": c.get("id"),
"platform": c.get("platform"),
"email": c.get("platform_email") or c.get("email"),
"status": c.get("status"),
}
for c in calendars
if isinstance(c, dict)
],
"callback_configured": bool(_calendar_regional_uri),
"public_api_base_url_set": bool(public_url),
"webhook_events": ["calendar.sync_events", "calendar.update"],
"next_step": (
"Complete Google Calendar V2 setup via Recall MCP "
"(start_calendar_integration_setup) with production_redirect_uri="
f"{public_url}/api/recall/calendar/callback"
if public_url
else "Set PUBLIC_API_BASE_URL to an HTTPS tunnel URL, then run "
"Recall MCP calendar setup."
"Register calendar webhooks on the same /webhooks/recall URL, "
"then complete Google Calendar V2 OAuth via Recall MCP."
),
}
}
)
async def api_recall_calendar_upcoming(request: Request) -> JSONResponse:
hours = int(request.query_params.get("hours", "24"))
try:
rows = await asyncio.to_thread(list_upcoming_calendar_meetings, hours=hours)
except RecallConfigError as exc:
return JSONResponse({"error": str(exc)}, status_code=503)
except Exception as exc: # noqa: BLE001
return JSONResponse({"error": str(exc)}, status_code=502)
return JSONResponse({"events": rows, "hours": hours})
async def api_recall_calendar_sync(request: Request) -> JSONResponse:
try:
body = await request.json()
except json.JSONDecodeError:
body = {}
calendar_id = str(body.get("calendar_id") or request.query_params.get("calendar_id") or "").strip()
if not calendar_id:
return JSONResponse({"error": "calendar_id is required"}, status_code=400)
try:
settings = load_recall_settings()
result = await asyncio.to_thread(
sync_calendar_events,
calendar_id,
updated_since=body.get("updated_at__gte"),
settings=settings,
)
except RecallConfigError as exc:
return JSONResponse({"error": str(exc)}, status_code=503)
except Exception as exc: # noqa: BLE001
return JSONResponse({"error": str(exc)}, status_code=502)
return JSONResponse(result)
async def api_meeting_live_transcript(request: Request) -> Response:
meeting_id = request.path_params["meeting_id"]
accept = request.headers.get("accept", "")
try:
from praisonai_tools.tools.meeting_tools import get_meeting
record = get_meeting.__wrapped__(meeting_id)
except ImportError:
return JSONResponse({"error": "meeting tools unavailable"}, status_code=503)
if "error" in record:
return JSONResponse(record, status_code=404)
if "text/event-stream" not in accept:
return JSONResponse(snapshot_for_api(meeting_id, record))
async def event_stream():
last_text = ""
while True:
try:
from praisonai_tools.tools.meeting_tools import get_meeting
record = get_meeting.__wrapped__(meeting_id)
except ImportError:
break
snap = snapshot_for_api(meeting_id, record)
text = snap.get("transcript") or ""
live = (snap.get("live_status") or "").lower()
if text != last_text:
yield f"data: {json.dumps(snap)}\n\n"
last_text = text
if live not in ("live", "joining", "waiting_room", "scheduled", ""):
break
await asyncio.to_thread(wait_for_update, meeting_id, 25.0)
return StreamingResponse(event_stream(), media_type="text/event-stream")
async def api_recall_config_status(_request: Request) -> JSONResponse:
try:
settings = load_recall_settings()
+112
View File
@@ -0,0 +1,112 @@
# Start meeting-agent (Part B) + Cloudflare tunnel for Recall webhooks.
# Usage: .\start_dev.ps1
# Optional: $env:TEST_MEETING_URL = "https://meet.google.com/..." ; .\start_dev.ps1 -ScheduleBot
param(
[switch]$ScheduleBot,
[string]$MeetingUrl = $env:TEST_MEETING_URL,
[int]$Port = 8000
)
$ErrorActionPreference = "Stop"
$Root = Split-Path -Parent $MyInvocation.MyCommand.Path
Set-Location $Root
function Load-DotEnv {
$envFile = Join-Path $Root ".env"
if (-not (Test-Path $envFile)) { return }
Get-Content $envFile | ForEach-Object {
$line = $_.Trim()
if (-not $line -or $line.StartsWith("#") -or $line -notmatch "=") { return }
$key, $value = $line -split "=", 2
if ($key -and -not [string]::IsNullOrWhiteSpace($key) -and -not (Get-Item "Env:$key" -ErrorAction SilentlyContinue)) {
Set-Item -Path "Env:$key" -Value $value.Trim()
}
}
}
function Test-PortListening([int]$p) {
try {
$r = Invoke-WebRequest -Uri "http://127.0.0.1:$p/api/recall/config" -UseBasicParsing -TimeoutSec 2
return $r.StatusCode -eq 200
} catch { return $false }
}
function Update-PublicUrl([string]$baseUrl) {
$envFile = Join-Path $Root ".env"
$content = Get-Content $envFile -Raw
if ($content -match "PUBLIC_API_BASE_URL=.*") {
$newContent = $content -replace "PUBLIC_API_BASE_URL=.*", "PUBLIC_API_BASE_URL=$baseUrl"
} else {
$newContent = $content.TrimEnd() + "`nPUBLIC_API_BASE_URL=$baseUrl`n"
}
Set-Content -Path $envFile -Value $newContent -NoNewline
$env:PUBLIC_API_BASE_URL = $baseUrl
Write-Host "Updated .env PUBLIC_API_BASE_URL=$baseUrl"
}
Load-DotEnv
$env:MEETING_AGENT_PORT = "$Port"
if (-not (Test-PortListening $Port)) {
Write-Host "Starting meeting-agent on port $Port..."
Start-Process -FilePath "py" -ArgumentList "-3.13", "app.py" -WorkingDirectory $Root -WindowStyle Minimized
$deadline = (Get-Date).AddSeconds(45)
while ((Get-Date) -lt $deadline) {
if (Test-PortListening $Port) { break }
Start-Sleep -Seconds 2
}
if (-not (Test-PortListening $Port)) {
throw "App did not become ready on http://127.0.0.1:$Port"
}
Write-Host "App ready: http://127.0.0.1:$Port"
} else {
Write-Host "App already running on port $Port"
}
$public = $env:PUBLIC_API_BASE_URL
try {
$null = Invoke-WebRequest -Uri "$public/api/recall/config" -UseBasicParsing -TimeoutSec 5
Write-Host "Tunnel OK: $public"
} catch {
Write-Host "Starting Cloudflare tunnel (quick tunnel - URL may change)..."
$log = Join-Path $Root ".tunnel.log"
$logErr = "${log}.err"
if (Test-Path $log) { Remove-Item $log -Force }
if (Test-Path $logErr) { Remove-Item $logErr -Force }
Start-Process -FilePath "cloudflared" -ArgumentList "tunnel", "--url", "http://127.0.0.1:$Port" `
-RedirectStandardOutput $log -RedirectStandardError $logErr -WindowStyle Minimized
$deadline = (Get-Date).AddSeconds(90)
$newUrl = $null
while ((Get-Date) -lt $deadline -and -not $newUrl) {
Start-Sleep -Seconds 2
foreach ($path in @($logErr, $log)) {
if (-not (Test-Path $path)) { continue }
$match = Select-String -Path $path -Pattern "https://[a-z0-9-]+\.trycloudflare\.com" | Select-Object -First 1
if ($match) {
$newUrl = $match.Matches[0].Value
break
}
}
}
if (-not $newUrl) { throw "Could not read tunnel URL from $logErr (or $log). Check cloudflared is installed." }
Update-PublicUrl $newUrl
Write-Host "Tunnel ready: $newUrl"
Write-Host "If Recall webhook URL changed, update it in Recall dashboard Webhooks"
}
$config = Invoke-RestMethod -Uri "http://127.0.0.1:$Port/api/recall/config"
Write-Host "Recall configured: $($config.configured)"
if ($ScheduleBot) {
if (-not $MeetingUrl) {
Write-Host "No TEST_MEETING_URL - skip bot schedule. Set env TEST_MEETING_URL or pass -MeetingUrl"
} else {
$body = @{ meeting_url = $MeetingUrl; title = "Part B live transcript test" } | ConvertTo-Json
$result = Invoke-RestMethod -Method Post -Uri "http://127.0.0.1:$Port/api/recall/bots" `
-ContentType "application/json" -Body $body
Write-Host "Bot scheduled:" ($result | ConvertTo-Json -Compress)
}
}
Write-Host "Done. Dashboard: http://127.0.0.1:$Port"
+16
View File
@@ -0,0 +1,16 @@
# Voice Agent — local secrets. Never commit.
OPENAI_API_KEY=
VOICE_AGENT_PORT=8001
# Voice telephony provider
VOICE_API_BASE=
VOICE_PRIVATE_API_KEY=
VOICE_SERVER_URL_SECRET=
VOICE_ASSISTANT_ID=
VOICE_PHONE_NUMBER_ID=
# Public HTTPS URL for provider webhook (Cloudflare tunnel)
# cloudflared tunnel --url http://127.0.0.1:8001
PUBLIC_API_BASE_URL=
+54
View File
@@ -0,0 +1,54 @@
# Voice Agent example — PraisonAIUI
Phone and web voice agents using an external telephony/voice platform for STT/TTS/calls and PraisonAI for tools + dashboard.
## Quick start
```powershell
cd examples/python/voice-agent
copy .env.example .env
# Fill VOICE_API_BASE, VOICE_PRIVATE_API_KEY, VOICE_ASSISTANT_ID, etc.
# Fully automatic (app + tunnel + provider config):
.\setup_all.ps1
# Or step by step:
.\start_dev.ps1
```
Dashboard: http://127.0.0.1:8001
## Provider setup
1. Create an **assistant** with tools `get_current_time` and `echo_message`
2. Set **webhook URL** to `{PUBLIC_API_BASE_URL}/webhooks/voice`
3. Set **webhook secret header** `X-Voice-Webhook-Secret` → same as `VOICE_SERVER_URL_SECRET` in `.env`
4. Add a phone number and copy `VOICE_PHONE_NUMBER_ID`
Or run `py -3.13 setup_voice_provider.py` after `start_dev.ps1`.
## API
| Route | Method | Purpose |
|-------|--------|---------|
| `/api/voice/calls` | POST | Outbound call `{ "customer_number": "+1..." }` |
| `/api/voice/calls` | GET | List stored calls |
| `/api/voice/calls/{id}` | GET | Call detail + transcript |
| `/webhooks/voice` | POST | Provider webhooks (tool-calls, transcript, end-of-call) |
| `/api/voice/config` | GET | Config status |
## Architecture
External platform handles voice pipeline (phone, STT, TTS, turn-taking). This app handles:
- **tool-calls** → PraisonAI-registered tools (`integrations/voice/tools.py`)
- **transcript** / **end-of-call-report** → SQLite store + dashboard pages
- **assistant-request** → returns `VOICE_ASSISTANT_ID`
Same pattern as `examples/python/meeting-agent` (Recall webhooks).
## Tests
```bash
pytest tests/unit/test_voice_agent.py -v
```
+208
View File
@@ -0,0 +1,208 @@
"""Voice Agent — PraisonAIUI example with telephony webhooks.
- Outbound calls → POST /api/voice/calls
- Provider webhooks → POST /webhooks/voice (tool-calls, transcript, end-of-call)
- Dashboard → Calls list + call detail + chat
Run:
cd examples/python/voice-agent
pip install httpx
copy .env.example .env
.\\start_dev.ps1
python app.py
"""
from __future__ import annotations
import asyncio
import os
import sys
from pathlib import Path
from typing import Any
from starlette.routing import Route
from voice_routes import (
api_voice_call_detail,
api_voice_config_status,
api_voice_create_call,
api_voice_list_calls,
webhook_voice,
)
import praisonaiui as aiui
from praisonaiui.server import create_app
_EXAMPLE_DIR = Path(__file__).resolve().parent
if str(_EXAMPLE_DIR) not in sys.path:
sys.path.insert(0, str(_EXAMPLE_DIR))
def _load_local_env() -> None:
env_path = _EXAMPLE_DIR / ".env"
if not env_path.is_file():
return
for raw in env_path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key, value = key.strip(), value.strip()
if key and key not in os.environ:
os.environ[key] = value
_load_local_env()
aiui.set_pages(["chat", "voice-calls", "call-detail", "config"])
aiui.set_style("dashboard")
_agent: Any | None = None
def get_agent():
global _agent
if _agent is not None:
return _agent
from integrations.voice.tools import echo_message, get_current_time, register_tool
from praisonaiagents import Agent
register_tool("get_current_time", get_current_time)
register_tool("echo_message", echo_message)
_agent = Agent(
name="Voice Assistant",
instructions=(
"You help with voice and phone agent demos. "
"Use tools when the caller asks for the time or wants a message echoed."
),
model="gpt-4o-mini",
tools=[get_current_time, echo_message],
)
from praisonaiagents.escalation.loop_guard import LoopGuard, LoopGuardConfig
max_turn_sec = float(os.getenv("VOICE_AGENT_LOOP_GUARD_MAX_SEC", "600"))
_agent._loop_guard = LoopGuard(
LoopGuardConfig(enabled=True, max_time_per_turn=max_turn_sec)
)
return _agent
def _demo_tool_reply(text: str) -> str | None:
"""Fast path for demo tool queries — no LLM required."""
from integrations.voice.tools import echo_message, get_current_time
lower = text.lower().strip()
if any(p in lower for p in ("what time", "current time", "tell me the time", "time is it")):
return str(get_current_time()["spoken"])
if lower.startswith("echo "):
return str(echo_message(text[5:])["spoken"])
if lower.startswith("repeat "):
return str(echo_message(text[7:])["spoken"])
return None
@aiui.reply
async def on_message(message: str):
"""Chat with the voice demo agent."""
text = str(message).strip()
if not text:
return
demo = _demo_tool_reply(text)
if demo is not None:
await aiui.say(demo)
return
if not os.getenv("OPENAI_API_KEY"):
await aiui.say(
"Set `OPENAI_API_KEY` for open-ended chat. "
'Demo without API key: "What time is it?" or "Echo hello world".'
)
return
await aiui.think("Running voice assistant...")
agent = get_agent()
response = await asyncio.to_thread(agent.chat, text)
await aiui.say(str(response))
def _list_calls() -> list[dict[str, Any]]:
from integrations.voice.store import VoiceCallStore
rows = VoiceCallStore.list_calls()
return [
{
"id": r["call_id"],
"status": r["status"],
"customer_number": r.get("customer_number") or "—",
"transcript_preview": (r.get("transcript") or "")[:120],
"updated_at": r.get("updated_at") or "",
}
for r in rows
]
def _pick_call(calls: list[dict[str, Any]]) -> dict[str, Any]:
live = next((c for c in calls if "progress" in (c.get("status") or "").lower()), None)
return live or (calls[0] if calls else {"id": "—", "status": "none", "customer_number": "—", "transcript_preview": ""})
@aiui.page("voice-calls", title="Voice calls", icon="📞", group="Voice", order=1)
async def voice_calls_page():
calls = _list_calls()
rows = [[c["id"], c["customer_number"], c["status"], c["updated_at"]] for c in calls]
return aiui.layout(
[
aiui.text("Outbound and inbound voice calls via telephony provider"),
aiui.table(
headers=["Call ID", "Customer", "Status", "Updated"],
rows=rows or [["—", "—", "—", "—"]],
),
aiui.alert(
"Start a call: POST /api/voice/calls with {\"customer_number\": \"+1...\"}. "
"Set provider webhook URL to {PUBLIC_API_BASE_URL}/webhooks/voice",
variant="info",
title="Setup",
),
]
)
@aiui.page("call-detail", title="Call detail", icon="🎙️", group="Voice", order=2)
async def call_detail_page():
from integrations.voice.store import VoiceCallStore
calls = _list_calls()
picked = _pick_call(calls)
record = VoiceCallStore.get_call(picked["id"]) if picked.get("id") and picked["id"] != "—" else None
transcript = (record or {}).get("transcript") or "(empty)"
summary = (record or {}).get("summary") or "(not available yet)"
status = (record or {}).get("status") or picked.get("status") or "unknown"
return aiui.layout(
[
aiui.text(f"Call {picked.get('id', '—')}"),
aiui.badge(status, variant="default"),
aiui.tabs(
[
{"label": "Transcript", "children": [aiui.code_block(transcript, language="text")]},
{"label": "Summary", "children": [aiui.text(summary)]},
]
),
]
)
app = create_app()
app.routes[0:0] = [
Route("/api/voice/calls", api_voice_create_call, methods=["POST"]),
Route("/api/voice/calls", api_voice_list_calls, methods=["GET"]),
Route("/api/voice/calls/{call_id}", api_voice_call_detail, methods=["GET"]),
Route("/api/voice/config", api_voice_config_status, methods=["GET"]),
Route("/webhooks/voice", webhook_voice, methods=["POST"]),
]
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("VOICE_AGENT_PORT", "8001"))
uvicorn.run(app, host=host, port=port)
@@ -0,0 +1 @@
"""Voice agent integrations."""
@@ -0,0 +1 @@
"""Voice telephony provider integration."""
@@ -0,0 +1,76 @@
"""Minimal voice provider REST client."""
from __future__ import annotations
import logging
from typing import Any
from integrations.voice.config import VoiceSettings
logger = logging.getLogger(__name__)
class VoiceAPIError(RuntimeError):
def __init__(self, status_code: int, detail: str):
super().__init__(detail)
self.status_code = status_code
self.detail = detail
class VoiceClient:
"""HTTP client for voice provider calls and assistants."""
def __init__(self, settings: VoiceSettings):
self._settings = settings
def _headers(self) -> dict[str, str]:
return {
"Authorization": f"Bearer {self._settings.api_key}",
"Accept": "application/json",
"Content-Type": "application/json",
}
def _request(
self,
method: str,
path: str,
*,
json_body: dict[str, Any] | None = None,
) -> dict[str, Any]:
import httpx # lazy: optional dependency for the example (pip install httpx)
url = f"{self._settings.api_base.rstrip('/')}{path}"
with httpx.Client(timeout=30.0) as client:
response = client.request(method, url, headers=self._headers(), json=json_body)
if response.status_code >= 400:
raise VoiceAPIError(response.status_code, response.text[:500] or response.reason_phrase)
if not response.content:
return {}
data = response.json()
return data if isinstance(data, dict) else {"data": data}
def create_call(
self,
*,
customer_number: str,
assistant_id: str | None = None,
phone_number_id: str | None = None,
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
assistant_id = assistant_id or self._settings.default_assistant_id
phone_number_id = phone_number_id or self._settings.default_phone_number_id
if not assistant_id:
raise VoiceAPIError(0, "assistant_id is required (set VOICE_ASSISTANT_ID or pass explicitly)")
if not phone_number_id:
raise VoiceAPIError(0, "phone_number_id is required (set VOICE_PHONE_NUMBER_ID or pass explicitly)")
body: dict[str, Any] = {
"assistantId": assistant_id,
"phoneNumberId": phone_number_id,
"customer": {"number": customer_number},
}
if metadata:
body["metadata"] = metadata
return self._request("POST", "/call", json_body=body)
def get_call(self, call_id: str) -> dict[str, Any]:
return self._request("GET", f"/call/{call_id}")
@@ -0,0 +1,65 @@
"""Voice provider configuration loaded from environment."""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
class VoiceConfigError(ValueError):
"""Missing or invalid voice provider configuration."""
@dataclass(frozen=True)
class VoiceSettings:
api_key: str
server_url_secret: str | None
public_api_base_url: str | None
default_assistant_id: str | None
default_phone_number_id: str | None
api_base: str
@property
def webhook_url(self) -> str | None:
if not self.public_api_base_url:
return None
return f"{self.public_api_base_url.rstrip('/')}/webhooks/voice"
@property
def configured(self) -> bool:
return bool(self.api_key and self.api_base)
def _read_dotenv(name: str) -> str | None:
root = Path(__file__).resolve().parents[2]
env_path = root / ".env"
if not env_path.is_file():
return None
prefix = f"{name}="
for raw in env_path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if line.startswith(prefix):
return line[len(prefix) :].strip() or None
return None
def _env(name: str) -> str:
return os.getenv(name, "").strip() or (_read_dotenv(name) or "")
def load_voice_settings(*, require_key: bool = True) -> VoiceSettings:
api_key = _env("VOICE_PRIVATE_API_KEY") or _env("VOICE_API_KEY")
api_base = _env("VOICE_API_BASE")
if require_key and not api_key:
raise VoiceConfigError("VOICE_PRIVATE_API_KEY (or VOICE_API_KEY) is required")
if require_key and not api_base:
raise VoiceConfigError("VOICE_API_BASE is required")
return VoiceSettings(
api_key=api_key,
server_url_secret=_env("VOICE_SERVER_URL_SECRET") or None,
public_api_base_url=_env("PUBLIC_API_BASE_URL") or None,
default_assistant_id=_env("VOICE_ASSISTANT_ID") or None,
default_phone_number_id=_env("VOICE_PHONE_NUMBER_ID") or None,
api_base=api_base.rstrip("/"),
)
@@ -0,0 +1,115 @@
"""Process voice provider webhook events."""
from __future__ import annotations
import json
import logging
from typing import Any
from integrations.voice.config import VoiceSettings
from integrations.voice.store import VoiceCallStore, webhook_event_key
from integrations.voice.tools import execute_tool
logger = logging.getLogger(__name__)
def _message(payload: dict[str, Any]) -> dict[str, Any]:
msg = payload.get("message")
return msg if isinstance(msg, dict) else payload
def _call_id(payload: dict[str, Any]) -> str | None:
msg = _message(payload)
call = msg.get("call")
if isinstance(call, dict) and call.get("id"):
return str(call["id"])
return None
def _customer_number(payload: dict[str, Any]) -> str | None:
msg = _message(payload)
call = msg.get("call") if isinstance(msg.get("call"), dict) else {}
customer = call.get("customer") if isinstance(call, dict) else {}
if isinstance(customer, dict) and customer.get("number"):
return str(customer["number"])
return None
def process_voice_webhook(
event_type: str, payload: dict[str, Any], *, settings: VoiceSettings
) -> dict[str, Any] | None:
"""Handle a verified provider webhook POST. Returns response body when required."""
event_key = webhook_event_key(event_type, payload)
if event_type not in ("tool-calls", "assistant-request", "transfer-destination-request", "knowledge-base-request"):
if not VoiceCallStore.mark_event_processed(event_key, event_type):
logger.info("Skipping duplicate voice event %s", event_type)
return None
msg = _message(payload)
call_id = _call_id(payload)
if event_type == "assistant-request":
assistant_id = settings.default_assistant_id
if not assistant_id:
return {"error": "No assistant configured on server"}
return {"assistantId": assistant_id}
if event_type == "tool-calls" and call_id:
tool_calls = msg.get("toolCallList") or []
results = []
for item in tool_calls:
if not isinstance(item, dict):
continue
tool_call_id = str(item.get("id") or "")
name = str(item.get("name") or "")
params = item.get("parameters") if isinstance(item.get("parameters"), dict) else {}
result = execute_tool(name, params)
results.append({"toolCallId": tool_call_id, "name": name, "result": result})
VoiceCallStore.upsert_call(call_id, status="in-progress", customer_number=_customer_number(payload))
return {"results": results}
if event_type == "status-update" and call_id:
status = str(msg.get("status") or "unknown")
VoiceCallStore.upsert_call(call_id, status=status, customer_number=_customer_number(payload))
return None
if event_type == "transcript" and call_id:
role = str(msg.get("role") or "unknown")
text = str(msg.get("transcript") or "").strip()
if text and str(msg.get("transcriptType") or "final") == "final":
VoiceCallStore.append_transcript_line(call_id, f"{role}: {text}")
return None
if event_type == "end-of-call-report" and call_id:
artifact = msg.get("artifact") if isinstance(msg.get("artifact"), dict) else {}
transcript = artifact.get("transcript") if isinstance(artifact, dict) else None
summary = str(msg.get("summary") or "")
ended = str(msg.get("endedReason") or "ended")
VoiceCallStore.upsert_call(
call_id,
status=f"ended ({ended})",
customer_number=_customer_number(payload),
transcript=str(transcript) if transcript else None,
summary=summary or None,
metadata={"artifact": artifact},
)
return None
if event_type == "conversation-update" and call_id:
messages = msg.get("messages") or []
lines = []
for item in messages:
if not isinstance(item, dict):
continue
role = item.get("role") or "unknown"
text = item.get("message") or item.get("content") or ""
if text:
lines.append(f"{role}: {text}")
if lines:
VoiceCallStore.upsert_call(
call_id, transcript="\n".join(lines), customer_number=_customer_number(payload)
)
return None
logger.debug("Unhandled voice event %s: %s", event_type, json.dumps(payload)[:300])
return None
@@ -0,0 +1,204 @@
"""Persist voice call state and webhook idempotency."""
from __future__ import annotations
import json
import os
import sqlite3
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
def _state_db_path() -> str:
env = os.getenv("PRAISONAI_VOICE_DIR")
base = Path(env) if env else Path.home() / ".praisonai" / "voice"
base.mkdir(parents=True, exist_ok=True)
return str(base / "voice_state.db")
def _connect() -> sqlite3.Connection:
conn = sqlite3.connect(_state_db_path())
conn.execute(
"""
CREATE TABLE IF NOT EXISTS voice_calls (
call_id TEXT PRIMARY KEY,
status TEXT NOT NULL,
customer_number TEXT,
transcript TEXT NOT NULL DEFAULT '',
summary TEXT NOT NULL DEFAULT '',
metadata_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS voice_webhook_events (
event_key TEXT PRIMARY KEY,
event_type TEXT NOT NULL,
processed_at TEXT NOT NULL
)
"""
)
return conn
def webhook_event_key(event_type: str, payload: dict[str, Any]) -> str:
message = payload.get("message") if isinstance(payload.get("message"), dict) else payload
call = (message or {}).get("call") if isinstance(message, dict) else {}
call_id = call.get("id") if isinstance(call, dict) else None
if isinstance(message, dict):
# Derive the key from the fields that actually distinguish one event from
# the next. A truncated JSON dump is unsafe here: when a large static field
# (e.g. ``call``) sorts before the discriminating field, successive events
# collapse to the same prefix and later ones get dropped as "duplicates"
# (e.g. a status-update stuck on "ringing").
if event_type == "transcript":
text = str(message.get("transcript") or "")
role = str(message.get("role") or "")
kind = str(message.get("transcriptType") or "")
return f"{event_type}:{call_id}:{role}:{kind}:{text}"
if event_type == "status-update":
return f"{event_type}:{call_id}:{message.get('status')}"
if event_type == "conversation-update":
messages = message.get("messages")
count = len(messages) if isinstance(messages, list) else 0
return f"{event_type}:{call_id}:{count}"
if event_type == "end-of-call-report":
return f"{event_type}:{call_id}:{message.get('endedReason')}"
if call_id:
return f"{event_type}:{call_id}:{json.dumps(message, sort_keys=True)[:200]}"
return f"{event_type}:{json.dumps(payload, sort_keys=True)[:300]}"
class VoiceCallStore:
@staticmethod
def mark_event_processed(event_key: str, event_type: str) -> bool:
now = datetime.now(timezone.utc).isoformat()
with closing(_connect()) as conn, conn:
row = conn.execute(
"SELECT 1 FROM voice_webhook_events WHERE event_key = ?",
(event_key,),
).fetchone()
if row:
return False
conn.execute(
"""
INSERT INTO voice_webhook_events (event_key, event_type, processed_at)
VALUES (?, ?, ?)
""",
(event_key, event_type, now),
)
return True
@staticmethod
def upsert_call(
call_id: str,
*,
status: str | None = None,
customer_number: str | None = None,
transcript: str | None = None,
summary: str | None = None,
metadata: dict[str, Any] | None = None,
) -> None:
now = datetime.now(timezone.utc).isoformat()
with closing(_connect()) as conn, conn:
row = conn.execute(
"SELECT transcript, summary, metadata_json, customer_number, status FROM voice_calls WHERE call_id = ?",
(call_id,),
).fetchone()
existing_transcript = row[0] if row else ""
existing_summary = row[1] if row else ""
existing_meta = json.loads(row[2]) if row and row[2] else {}
existing_number = row[3] if row else None
existing_status = row[4] if row else None
# Preserve prior status when the caller doesn't supply one (e.g. transcript
# / conversation-update events) so partial updates don't clobber live state.
final_status = status if status is not None else (existing_status or "unknown")
merged_meta = {**existing_meta, **(metadata or {})}
conn.execute(
"""
INSERT INTO voice_calls (
call_id, status, customer_number, transcript, summary, metadata_json, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(call_id) DO UPDATE SET
status = excluded.status,
customer_number = COALESCE(excluded.customer_number, voice_calls.customer_number),
transcript = excluded.transcript,
summary = COALESCE(excluded.summary, voice_calls.summary),
metadata_json = excluded.metadata_json,
updated_at = excluded.updated_at
""",
(
call_id,
final_status,
customer_number or existing_number,
transcript if transcript is not None else existing_transcript,
summary if summary is not None else existing_summary,
json.dumps(merged_meta, sort_keys=True),
now,
now,
),
)
@staticmethod
def append_transcript_line(call_id: str, line: str) -> str:
with closing(_connect()) as conn:
row = conn.execute(
"SELECT transcript FROM voice_calls WHERE call_id = ?",
(call_id,),
).fetchone()
existing = row[0] if row else ""
full = f"{existing}\n{line}".strip() if existing else line
VoiceCallStore.upsert_call(call_id, transcript=full)
return full
@staticmethod
def list_calls(limit: int = 50) -> list[dict[str, Any]]:
with closing(_connect()) as conn:
rows = conn.execute(
"""
SELECT call_id, status, customer_number, transcript, summary, metadata_json, updated_at
FROM voice_calls ORDER BY updated_at DESC LIMIT ?
""",
(max(1, limit),),
).fetchall()
out: list[dict[str, Any]] = []
for row in rows:
out.append(
{
"call_id": row[0],
"status": row[1],
"customer_number": row[2],
"transcript": row[3],
"summary": row[4],
"metadata": json.loads(row[5] or "{}"),
"updated_at": row[6],
}
)
return out
@staticmethod
def get_call(call_id: str) -> dict[str, Any] | None:
with closing(_connect()) as conn:
row = conn.execute(
"""
SELECT call_id, status, customer_number, transcript, summary, metadata_json, updated_at
FROM voice_calls WHERE call_id = ?
""",
(call_id,),
).fetchone()
if not row:
return None
return {
"call_id": row[0],
"status": row[1],
"customer_number": row[2],
"transcript": row[3],
"summary": row[4],
"metadata": json.loads(row[5] or "{}"),
"updated_at": row[6],
}
@@ -0,0 +1,49 @@
"""Tools callable from voice provider ``tool-calls`` webhook events."""
from __future__ import annotations
import json
from datetime import datetime, timezone
from typing import Any, Callable
def get_current_time() -> dict[str, Any]:
"""Return current UTC time for voice agent demos."""
now = datetime.now(timezone.utc).isoformat()
return {"utc": now, "spoken": f"The current UTC time is {now}"}
def echo_message(message: str = "") -> dict[str, Any]:
"""Echo a message back to the caller."""
text = message.strip() or "nothing"
return {"message": text, "spoken": f"You said {text}"}
_TOOL_REGISTRY: dict[str, Callable[..., Any]] = {
"get_current_time": get_current_time,
"echo_message": echo_message,
}
def register_tool(name: str, fn: Callable[..., Any]) -> None:
_TOOL_REGISTRY[name] = fn
def execute_tool(name: str, parameters: dict[str, Any] | None) -> str:
"""Run a registered tool and return a JSON string for the provider."""
fn = _TOOL_REGISTRY.get(name)
if fn is None:
return json.dumps({"error": f"Unknown tool: {name}"})
try:
result = fn(**(parameters or {}))
if isinstance(result, str):
return result
return json.dumps(result)
except TypeError:
try:
result = fn(parameters or {})
return json.dumps(result)
except Exception as exc: # noqa: BLE001
return json.dumps({"error": str(exc)})
except Exception as exc: # noqa: BLE001
return json.dumps({"error": str(exc)})
@@ -0,0 +1,21 @@
"""Verify incoming voice provider webhook requests."""
from __future__ import annotations
import hmac
_SECRET_HEADERS = ("x-voice-webhook-secret", "x-webhook-secret")
class VerificationError(ValueError):
"""Webhook secret verification failed."""
def verify_webhook_request(*, secret: str | None, headers: dict[str, str]) -> None:
"""Validate webhook secret header when ``VOICE_SERVER_URL_SECRET`` is configured."""
if not secret:
return
lowered = {k.lower(): v for k, v in headers.items()}
incoming = next((lowered[name] for name in _SECRET_HEADERS if lowered.get(name)), None)
if not incoming or not hmac.compare_digest(incoming, secret):
raise VerificationError("Invalid or missing webhook secret header")
+11
View File
@@ -0,0 +1,11 @@
# One command: start app + tunnel + configure voice provider automatically.
Set-Location $PSScriptRoot
if (-not (Test-Path .env)) { Copy-Item .env.example .env; Write-Host "Created .env — add VOICE_* keys" ; exit 1 }
.\start_dev.ps1
if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
Write-Host ""
Write-Host "Configuring voice provider via API..." -ForegroundColor Cyan
py -3.13 setup_voice_provider.py
exit $LASTEXITCODE
@@ -0,0 +1,246 @@
"""Configure voice provider assistant: tools + webhook URL + auth header.
Usage:
cd examples/python/voice-agent
py -3.13 setup_voice_provider.py
"""
from __future__ import annotations
import os
import uuid
from pathlib import Path
from typing import Any
import httpx
_ROOT = Path(__file__).resolve().parent
ENV_PATH = _ROOT / ".env"
WEBHOOK_SECRET_HEADER = "X-Voice-Webhook-Secret"
def _load_env() -> None:
if not ENV_PATH.is_file():
return
for raw in ENV_PATH.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key, value = key.strip(), value.strip()
if key and key not in os.environ:
os.environ[key] = value
def _write_env_value(key: str, value: str) -> None:
lines = ENV_PATH.read_text(encoding="utf-8").splitlines() if ENV_PATH.is_file() else []
updated = False
new_lines: list[str] = []
for line in lines:
if line.strip().startswith(f"{key}="):
new_lines.append(f"{key}={value}")
updated = True
else:
new_lines.append(line)
if not updated:
new_lines.append(f"{key}={value}")
ENV_PATH.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
os.environ[key] = value
def _api_key() -> str:
return os.getenv("VOICE_PRIVATE_API_KEY", "").strip() or os.getenv("VOICE_API_KEY", "").strip()
def _api_base() -> str:
return os.getenv("VOICE_API_BASE", "").strip().rstrip("/")
def _request(
method: str,
path: str,
*,
api_key: str,
api_base: str,
body: dict | None = None,
) -> dict | list | None:
url = f"{api_base}{path}"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
response = httpx.request(method, url, headers=headers, json=body, timeout=30.0)
if response.status_code >= 400:
print(f"ERROR {method} {path} -> {response.status_code}")
print(response.text[:1500])
return None
if not response.content:
return {}
return response.json()
def _tool_specs() -> list[dict[str, Any]]:
return [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "Returns the current UTC time when the caller asks for the time.",
"parameters": {"type": "object", "properties": {}, "required": []},
},
"async": False,
},
{
"type": "function",
"function": {
"name": "echo_message",
"description": "Repeats back a message the caller provides.",
"parameters": {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "The message to repeat back to the caller.",
}
},
"required": ["message"],
},
},
"async": False,
},
]
def ensure_server_secret() -> str:
secret = os.getenv("VOICE_SERVER_URL_SECRET", "").strip()
if not secret:
secret = str(uuid.uuid4())
_write_env_value("VOICE_SERVER_URL_SECRET", secret)
print("Generated VOICE_SERVER_URL_SECRET")
return secret
def ensure_tools(api_key: str, api_base: str, assistant_id: str) -> list[str]:
existing = _request("GET", "/tool", api_key=api_key, api_base=api_base)
by_name: dict[str, str] = {}
if isinstance(existing, list):
for item in existing:
fn = item.get("function") if isinstance(item, dict) else None
if isinstance(fn, dict) and fn.get("name") and item.get("id"):
by_name[str(fn["name"])] = str(item["id"])
tool_ids: list[str] = []
for spec in _tool_specs():
name = spec["function"]["name"]
if name in by_name:
tool_id = by_name[name]
print(f"Reusing tool {name}: {tool_id}")
else:
created = _request("POST", "/tool", api_key=api_key, api_base=api_base, body=spec)
if not isinstance(created, dict) or not created.get("id"):
raise RuntimeError(f"Failed to create tool {name}")
tool_id = str(created["id"])
print(f"Created tool {name}: {tool_id}")
tool_ids.append(tool_id)
assistant = _request("GET", f"/assistant/{assistant_id}", api_key=api_key, api_base=api_base)
if not isinstance(assistant, dict):
raise RuntimeError("Could not load assistant")
model = dict(assistant.get("model") or {})
if not model.get("provider"):
raise RuntimeError("Assistant has no model.provider — configure a model in the provider dashboard first")
existing_ids = [str(x) for x in (model.get("toolIds") or [])]
model["toolIds"] = list(dict.fromkeys(existing_ids + tool_ids))
messages = list(model.get("messages") or [])
system_hint = (
"You are a helpful voice assistant. When the caller asks for the time, "
"call get_current_time. When they ask you to repeat something, call echo_message."
)
if not messages:
model["messages"] = [{"role": "system", "content": system_hint}]
else:
for msg in messages:
if isinstance(msg, dict) and msg.get("role") == "system":
content = str(msg.get("content") or "")
if "get_current_time" not in content:
msg["content"] = f"{content.rstrip()}\n\n{system_hint}"
break
else:
messages.insert(0, {"role": "system", "content": system_hint})
model["messages"] = messages
patched = _request(
"PATCH",
f"/assistant/{assistant_id}",
api_key=api_key,
api_base=api_base,
body={"model": model},
)
if patched is None:
raise RuntimeError("Failed to attach tools to assistant")
print("Attached tool IDs:", model["toolIds"])
return tool_ids
def ensure_webhook(api_key: str, api_base: str, assistant_id: str, webhook_url: str, secret: str) -> None:
assistant = _request("GET", f"/assistant/{assistant_id}", api_key=api_key, api_base=api_base)
if not isinstance(assistant, dict):
raise RuntimeError("Could not load assistant for webhook setup")
server = dict(assistant.get("server") or {})
server["url"] = webhook_url.rstrip("/")
server["headers"] = {WEBHOOK_SECRET_HEADER: secret}
server.pop("credentialId", None)
server.pop("secret", None)
patched = _request(
"PATCH",
f"/assistant/{assistant_id}",
api_key=api_key,
api_base=api_base,
body={"server": server},
)
if patched is None:
raise RuntimeError("Failed to configure assistant server URL")
print(f"Server URL: {webhook_url}")
print(f"Webhook auth header: {WEBHOOK_SECRET_HEADER}")
def main() -> int:
_load_env()
api_key = _api_key()
api_base = _api_base()
assistant_id = os.getenv("VOICE_ASSISTANT_ID", "").strip()
public_base = os.getenv("PUBLIC_API_BASE_URL", "").strip().rstrip("/")
if not api_key:
print("Set VOICE_PRIVATE_API_KEY in .env")
return 1
if not api_base:
print("Set VOICE_API_BASE in .env")
return 1
if not assistant_id:
print("Set VOICE_ASSISTANT_ID in .env")
return 1
if not public_base:
print("Set PUBLIC_API_BASE_URL in .env (run .\\start_dev.ps1 first)")
return 1
secret = ensure_server_secret()
webhook_url = f"{public_base}/webhooks/voice"
try:
ensure_tools(api_key, api_base, assistant_id)
ensure_webhook(api_key, api_base, assistant_id, webhook_url, secret)
except RuntimeError as exc:
print(f"Setup failed: {exc}")
return 1
print("Voice provider auto-setup complete.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+221
View File
@@ -0,0 +1,221 @@
# Start voice-agent + Cloudflare tunnel for provider webhook URL.
# Usage: .\start_dev.ps1
# .\start_dev.ps1 -Restart # kill old app and reload code
param(
[int]$Port = 8001,
[switch]$Restart
)
$ErrorActionPreference = "Stop"
$Root = Split-Path -Parent $MyInvocation.MyCommand.Path
Set-Location $Root
function Load-DotEnv {
$envFile = Join-Path $Root ".env"
if (-not (Test-Path $envFile)) { return }
Get-Content $envFile | ForEach-Object {
$line = $_.Trim()
if (-not $line -or $line.StartsWith("#") -or $line -notmatch "=") { return }
$key, $value = $line -split "=", 2
if ($key -and -not [string]::IsNullOrWhiteSpace($key) -and -not (Get-Item "Env:$key" -ErrorAction SilentlyContinue)) {
Set-Item -Path "Env:$key" -Value $value.Trim()
}
}
}
function Test-PortListening([int]$p) {
foreach ($path in @("/health/live", "/health", "/api/voice/config")) {
try {
$r = Invoke-WebRequest -Uri "http://127.0.0.1:$p$path" -UseBasicParsing -TimeoutSec 15
if ($r.StatusCode -eq 200) { return $true }
} catch { continue }
}
return $false
}
function Stop-PortProcess([int]$p) {
try {
$conns = Get-NetTCPConnection -LocalPort $p -State Listen -ErrorAction SilentlyContinue
foreach ($conn in $conns) {
if ($conn.OwningProcess) {
Stop-Process -Id $conn.OwningProcess -Force -ErrorAction SilentlyContinue
}
}
if ($conns) {
Start-Sleep -Seconds 2
Write-Host "Stopped previous process on port $p"
}
} catch { }
}
function Clear-LogFile([string]$path) {
if (-not (Test-Path $path)) { return }
try {
Remove-Item $path -Force -ErrorAction Stop
} catch {
try { Clear-Content $path -ErrorAction SilentlyContinue } catch { }
}
}
function Start-VoiceApp([int]$p) {
Write-Host "Starting voice-agent on port $p..."
Stop-PortProcess $p
$appLog = Join-Path $Root ".app.log"
$appErr = Join-Path $Root ".app.err"
Clear-LogFile $appLog
Clear-LogFile $appErr
Start-Process -FilePath "py" -ArgumentList "-3.13", "app.py" -WorkingDirectory $Root -WindowStyle Minimized `
-RedirectStandardOutput $appLog -RedirectStandardError $appErr
$deadline = (Get-Date).AddSeconds(120)
while ((Get-Date) -lt $deadline) {
if (Test-PortListening $p) { break }
Start-Sleep -Seconds 3
}
if (-not (Test-PortListening $p)) {
$errText = ""
if (Test-Path $appErr) { $errText = Get-Content $appErr -Raw -ErrorAction SilentlyContinue }
if ($errText -match "Uvicorn running") {
Write-Host "Server process up — waiting for HTTP endpoints..."
Start-Sleep -Seconds 10
}
}
if (-not (Test-PortListening $p)) {
if (Test-Path $appErr) {
Write-Host "App startup log:" -ForegroundColor Red
Get-Content $appErr -Tail 20 | ForEach-Object { Write-Host $_ }
}
throw "App did not become ready on http://127.0.0.1:$p"
}
Write-Host "App ready: http://127.0.0.1:$p"
}
function Update-PublicUrl([string]$baseUrl) {
$envFile = Join-Path $Root ".env"
$content = Get-Content $envFile -Raw
if ($content -match "PUBLIC_API_BASE_URL=.*") {
$newContent = $content -replace "PUBLIC_API_BASE_URL=.*", "PUBLIC_API_BASE_URL=$baseUrl"
} else {
$newContent = $content.TrimEnd() + "`nPUBLIC_API_BASE_URL=$baseUrl`n"
}
Set-Content -Path $envFile -Value $newContent -NoNewline
$env:PUBLIC_API_BASE_URL = $baseUrl
Write-Host "Updated .env PUBLIC_API_BASE_URL=$baseUrl"
}
function Ensure-ServerSecret {
$envFile = Join-Path $Root ".env"
if (-not (Test-Path $envFile)) { return }
$lines = Get-Content $envFile
$secret = $env:VOICE_SERVER_URL_SECRET
foreach ($line in $lines) {
if ($line -match '^\s*VOICE_SERVER_URL_SECRET=(.+)$') {
$secret = $Matches[1].Trim()
break
}
}
if (-not $secret) {
$secret = [guid]::NewGuid().ToString()
$updated = $false
$newLines = foreach ($line in $lines) {
if ($line -match '^\s*VOICE_SERVER_URL_SECRET=') {
$updated = $true
"VOICE_SERVER_URL_SECRET=$secret"
} else { $line }
}
if (-not $updated) { $newLines += "VOICE_SERVER_URL_SECRET=$secret" }
Set-Content -Path $envFile -Value ($newLines -join "`n")
Write-Host "Generated VOICE_SERVER_URL_SECRET in .env"
}
$env:VOICE_SERVER_URL_SECRET = $secret
}
function Show-ProviderSetupSteps([string]$webhookUrl, [string]$secret) {
Write-Host ""
Write-Host "=== Provider webhook setup (Assistant -> Advanced -> Webhook Server) ===" -ForegroundColor Cyan
Write-Host "Webhook URL:"
Write-Host " $webhookUrl"
Write-Host ""
Write-Host "Secret header:"
Write-Host " Name: X-Voice-Webhook-Secret"
Write-Host " Value: $secret"
Write-Host "================================================================" -ForegroundColor Cyan
}
Load-DotEnv
Ensure-ServerSecret
$env:VOICE_AGENT_PORT = "$Port"
$needsRestart = $Restart.IsPresent
if (Test-PortListening $Port) {
if ($needsRestart) {
Stop-PortProcess $Port
} else {
try {
$cfg = Invoke-RestMethod -Uri "http://127.0.0.1:$Port/api/voice/config" -TimeoutSec 3
if (-not $cfg.webhook_url -and $env:PUBLIC_API_BASE_URL) {
Write-Host "App running but missing webhook URL — restarting to reload .env"
$needsRestart = $true
Stop-PortProcess $Port
} else {
Write-Host "App already running on port $Port"
}
} catch {
Write-Host "App already running on port $Port"
}
}
}
if ($needsRestart -or -not (Test-PortListening $Port)) {
Start-VoiceApp $Port
}
$public = $env:PUBLIC_API_BASE_URL
try {
$null = Invoke-WebRequest -Uri "$public/api/voice/config" -UseBasicParsing -TimeoutSec 5
Write-Host "Tunnel OK: $public"
} catch {
Write-Host "Starting Cloudflare tunnel..."
$log = Join-Path $Root ".tunnel.log"
$logErr = "${log}.err"
if (Test-Path $log) { Remove-Item $log -Force }
if (Test-Path $logErr) { Remove-Item $logErr -Force }
Start-Process -FilePath "cloudflared" -ArgumentList "tunnel", "--url", "http://127.0.0.1:$Port" `
-RedirectStandardOutput $log -RedirectStandardError $logErr -WindowStyle Minimized
$deadline = (Get-Date).AddSeconds(90)
$newUrl = $null
while ((Get-Date) -lt $deadline -and -not $newUrl) {
Start-Sleep -Seconds 2
foreach ($path in @($logErr, $log)) {
if (-not (Test-Path $path)) { continue }
$match = Select-String -Path $path -Pattern "https://[a-z0-9-]+\.trycloudflare\.com" | Select-Object -First 1
if ($match) {
$newUrl = $match.Matches[0].Value
break
}
}
}
if (-not $newUrl) { throw "Could not read tunnel URL from $logErr" }
Update-PublicUrl $newUrl
Write-Host "Tunnel ready: $newUrl"
if ($needsRestart -or -not (Test-PortListening $Port)) {
Stop-PortProcess $Port
Start-VoiceApp $Port
}
}
$config = Invoke-RestMethod -Uri "http://127.0.0.1:$Port/api/voice/config"
Write-Host "Voice configured: $($config.configured)"
Write-Host "Webhook URL: $($config.webhook_url)"
$webhook = if ($config.webhook_url) { $config.webhook_url } else { "$($env:PUBLIC_API_BASE_URL)/webhooks/voice" }
Write-Host ""
Write-Host "Configuring voice provider via API..." -ForegroundColor Cyan
py -3.13 setup_voice_provider.py
if ($LASTEXITCODE -eq 0) {
Write-Host "Voice provider auto-configured (tools + webhook URL + auth)." -ForegroundColor Green
} else {
Show-ProviderSetupSteps -webhookUrl $webhook -secret $env:VOICE_SERVER_URL_SECRET
}
Write-Host "Done. Dashboard: http://127.0.0.1:$Port"
+113
View File
@@ -0,0 +1,113 @@
"""Voice agent HTTP routes."""
from __future__ import annotations
import asyncio
import json
import logging
from integrations.voice.client import VoiceClient
from integrations.voice.config import VoiceConfigError, load_voice_settings
from integrations.voice.processor import process_voice_webhook
from integrations.voice.store import VoiceCallStore
from integrations.voice.verify import VerificationError, verify_webhook_request
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse, Response
logger = logging.getLogger(__name__)
# Hold strong references to background tasks so the event loop (which only keeps
# weak references) cannot garbage-collect them mid-flight and drop a webhook event.
_background_tasks: set[asyncio.Task] = set()
async def api_voice_config_status(_request: Request) -> JSONResponse:
try:
settings = load_voice_settings()
configured = True
error = None
except VoiceConfigError as exc:
settings = None
configured = False
error = str(exc)
return JSONResponse(
{
"configured": configured,
"error": error,
"webhook_url": settings.webhook_url if settings else None,
"default_assistant_id": settings.default_assistant_id if settings else None,
}
)
async def api_voice_create_call(request: Request) -> JSONResponse:
try:
body = await request.json()
except json.JSONDecodeError:
return JSONResponse({"error": "invalid JSON body"}, status_code=400)
number = str(body.get("customer_number") or body.get("phone_number") or "").strip()
if not number:
return JSONResponse({"error": "customer_number is required"}, status_code=400)
try:
settings = load_voice_settings()
client = VoiceClient(settings)
result = await asyncio.to_thread(
client.create_call,
customer_number=number,
assistant_id=body.get("assistant_id"),
phone_number_id=body.get("phone_number_id"),
metadata=body.get("metadata") if isinstance(body.get("metadata"), dict) else None,
)
except VoiceConfigError as exc:
return JSONResponse({"error": str(exc)}, status_code=503)
except Exception as exc: # noqa: BLE001
return JSONResponse({"error": str(exc)}, status_code=502)
call_id = result.get("id")
if call_id:
VoiceCallStore.upsert_call(str(call_id), status="scheduled", customer_number=number)
return JSONResponse(result, status_code=201)
async def api_voice_list_calls(_request: Request) -> JSONResponse:
return JSONResponse({"calls": VoiceCallStore.list_calls()})
async def api_voice_call_detail(request: Request) -> JSONResponse:
call_id = request.path_params["call_id"]
record = VoiceCallStore.get_call(call_id)
if not record:
return JSONResponse({"error": "not found"}, status_code=404)
return JSONResponse(record)
async def webhook_voice(request: Request) -> Response:
raw = (await request.body()).decode("utf-8")
try:
settings = load_voice_settings(require_key=False)
except VoiceConfigError as exc:
return JSONResponse({"error": str(exc)}, status_code=503)
headers = {k: v for k, v in request.headers.items()}
try:
verify_webhook_request(secret=settings.server_url_secret, headers=headers)
except VerificationError:
return JSONResponse({"error": "invalid secret"}, status_code=401)
try:
payload = json.loads(raw) if raw else {}
except json.JSONDecodeError:
return JSONResponse({"error": "invalid JSON"}, status_code=400)
msg = payload.get("message") if isinstance(payload.get("message"), dict) else payload
event_type = str((msg or {}).get("type") or payload.get("type") or "unknown")
if event_type in ("tool-calls", "assistant-request", "transfer-destination-request", "knowledge-base-request"):
result = await asyncio.to_thread(process_voice_webhook, event_type, payload, settings=settings)
return JSONResponse(result or {})
task = asyncio.create_task(
asyncio.to_thread(process_voice_webhook, event_type, payload, settings=settings)
)
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
return PlainTextResponse("ok", status_code=200)
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "aiui"
version = "0.3.142"
version = "0.3.143"
description = "YAML-driven website generator - CLI and compiler"
readme = "README.md"
license = {text = "MIT"}
+1 -1
View File
@@ -1,3 +1,3 @@
"""Version information."""
__version__ = "0.3.142"
__version__ = "0.3.143"
+27
View File
@@ -0,0 +1,27 @@
"""Shared pytest hooks for PraisonAIUI unit tests."""
from __future__ import annotations
import sys
from pathlib import Path
_MEETING_AGENT_ROOT = (
Path(__file__).resolve().parents[2] / "examples" / "python" / "meeting-agent"
)
def _ensure_meeting_agent_integrations() -> None:
"""Prefer meeting-agent ``integrations/`` over ``tests/unit/integrations/``."""
root = str(_MEETING_AGENT_ROOT)
if sys.path[:1] != [root]:
if root in sys.path:
sys.path.remove(root)
sys.path.insert(0, root)
for name in list(sys.modules):
if name == "integrations" or name.startswith("integrations."):
del sys.modules[name]
def pytest_runtest_setup(item) -> None:
if "meeting_agent" in item.nodeid.replace("\\", "/"):
_ensure_meeting_agent_integrations()
+141
View File
@@ -0,0 +1,141 @@
"""Unit tests for Meeting Agent Phase 2 Part B (live transcript + calendar)."""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
_ROOT = Path(__file__).resolve().parents[2] / "examples" / "python" / "meeting-agent"
if str(_ROOT) not in sys.path:
sys.path.insert(0, str(_ROOT))
def _load(name: str, path: Path):
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
@pytest.fixture(autouse=True)
def part_b_env(monkeypatch, tmp_path):
monkeypatch.setenv("RECALL_REGION", "eu-central-1")
monkeypatch.setenv("RECALL_API_KEY", "test-recall-key")
monkeypatch.setenv("RECALL_WEBHOOK_VERIFICATION_SECRET", "whsec_test")
monkeypatch.setenv("RECALL_WORKSPACE_ID", "33a035c7-03c1-4ec9-b752-18cdfbbefc42")
monkeypatch.setenv("PUBLIC_API_BASE_URL", "https://dev.example.test")
monkeypatch.setenv("PRAISONAI_MEETINGS_DIR", str(tmp_path))
monkeypatch.setenv("PRAISONAI_MEETINGS_DB", str(tmp_path / "meetings.db"))
monkeypatch.setenv("OPENAI_API_KEY", "test-openai")
def test_transcript_data_to_line():
tx = _load("part_b_transcript", _ROOT / "integrations" / "recall" / "transcript.py")
payload = {
"data": {
"data": {
"speaker": "Alice",
"words": [{"text": "Hello"}, {"text": "team"}],
}
}
}
assert tx.transcript_data_to_line(payload) == "Alice: Hello team"
def test_live_hub_publish_and_snapshot():
live = _load("part_b_live", _ROOT / "integrations" / "recall" / "live.py")
live.publish_live_chunk("m-live", line="Alice: Hi", full_text="Alice: Hi", live_status="live")
snap = live.snapshot_for_api("m-live", {"metadata": {}})
assert snap["transcript"] == "Alice: Hi"
assert snap["live_status"] == "live"
def test_create_bot_includes_realtime_config():
client_mod = _load("part_b_client", _ROOT / "integrations" / "recall" / "client.py")
config_mod = _load("part_b_config", _ROOT / "integrations" / "recall" / "config.py")
settings = config_mod.load_recall_settings()
client = client_mod.RecallClient(settings)
client._request = MagicMock(return_value={"id": "bot-1"}) # noqa: SLF001
client.create_bot(
meeting_url="https://meet.google.com/abc-defg-hij",
meeting_id="m1",
title="Standup",
)
body = client._request.call_args.kwargs["json_body"] # noqa: SLF001
assert "recording_config" in body
assert body["recording_config"]["realtime_endpoints"][0]["events"] == [
"transcript.data",
"transcript.partial_data",
]
def test_calendar_eligibility():
cal = _load("part_b_calendar", _ROOT / "integrations" / "recall" / "calendar_service.py")
assert cal.is_eligible_for_auto_record(
{
"is_deleted": False,
"meeting_url": "https://zoom.us/j/1",
"end_time": "2099-01-01T00:00:00+00:00",
}
)
assert not cal.is_eligible_for_auto_record({"is_deleted": True, "meeting_url": "https://zoom.us/j/1"})
def test_transcript_webhook_keys_are_unique_per_utterance():
store = _load("part_b_store", _ROOT / "integrations" / "recall" / "store.py")
bot = "bot-1"
p1 = {
"data": {
"bot": {"id": bot},
"data": {"speaker": "Alice", "words": [{"text": "hello", "start": 1.0, "end": 1.5}]},
}
}
p2 = {
"data": {
"bot": {"id": bot},
"data": {"speaker": "Alice", "words": [{"text": "world", "start": 2.0, "end": 2.5}]},
}
}
k1 = store.webhook_event_key("transcript.data", p1)
k2 = store.webhook_event_key("transcript.data", p2)
assert k1 != k2
def test_upsert_live_transcript_replaces_partial():
pytest.importorskip("praisonai_tools") # external PraisonAI-Tools dep, optional in CI
processor = _load("part_b_processor2", _ROOT / "integrations" / "recall" / "processor.py")
with patch("meeting_integrations.recall.live.publish_live_chunk") as publish:
with patch("praisonai_tools.tools.meeting_tools.get_meeting") as get_m:
get_m.__wrapped__ = MagicMock(
return_value={"metadata": {"transcript": "Alice: hel"}}
)
with patch.object(processor, "_merge_metadata"):
processor._upsert_live_transcript("m1", "Alice: hello", partial=True)
assert publish.call_args.kwargs["full_text"] == "Alice: hello"
def test_process_transcript_data_webhook():
processor = _load("part_b_processor", _ROOT / "integrations" / "recall" / "processor.py")
config_mod = _load("part_b_config2", _ROOT / "integrations" / "recall" / "config.py")
settings = config_mod.load_recall_settings()
with patch.object(processor, "_append_live_transcript") as append:
processor.process_recall_webhook(
"transcript.data",
{
"data": {
"bot": {"id": "b1", "metadata": {"meeting_id": "m1"}},
"data": {"speaker": "Bob", "text": "Testing live"},
}
},
settings=settings,
client=MagicMock(),
)
append.assert_called_once()
assert append.call_args[0][1] == "Bob: Testing live"
+1 -2
View File
@@ -9,8 +9,7 @@ from unittest.mock import patch
import pytest
# Meeting tools live in the external PraisonAI-Tools package (agent-callable
# integrations layer). Skip when it is not installed in this environment.
# Meeting tools live in the external PraisonAI-Tools package; skip when unavailable in CI.
pytest.importorskip("praisonai_tools")
_PIPELINE = (
+1 -3
View File
@@ -77,9 +77,7 @@ def test_verify_request_rejects_bad_signature():
def test_schedule_bot_persists_intent_before_api():
# schedule_recall_bot persists meeting intent via PraisonAI-Tools meeting
# tools; skip when that external package is not installed here.
pytest.importorskip("praisonai_tools")
pytest.importorskip("praisonai_tools") # external PraisonAI-Tools dep, optional in CI
service = _load("recall_service", _ROOT / "meeting_integrations" / "recall" / "service.py")
client = MagicMock()
client.create_bot.return_value = {"id": "bot-abc"}
+5
View File
@@ -55,6 +55,11 @@ class TestHealthEndpoint:
"""Test liveness endpoint returns quickly."""
import time
# Warm up once so the timed call measures steady-state latency rather
# than one-time TestClient/route initialization (which can exceed the
# threshold on slow CI runners and cause flaky failures).
client.get("/health/live")
start = time.time()
response = client.get("/health/live")
duration = time.time() - start
+208
View File
@@ -0,0 +1,208 @@
"""Unit tests for Voice Agent integration.
The example lives in ``examples/python/voice-agent`` and ships its own
``integrations.voice`` package. The wider test-suite also owns a top-level
``integrations`` package (``tests/unit/integrations``), so importing the
example's modules requires temporarily giving the example's directory
priority on ``sys.path`` and restoring the original ``integrations`` package
afterwards — otherwise the cached test package shadows ``integrations.voice``.
"""
from __future__ import annotations
import contextlib
import importlib
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
_ROOT = Path(__file__).resolve().parents[2] / "examples" / "python" / "voice-agent"
_INTEGRATIONS_PREFIX = ("integrations", "integrations.")
@contextlib.contextmanager
def _voice_modules():
"""Import the example's ``integrations.voice`` package in isolation.
Saves and clears any already-imported ``integrations*`` modules, puts the
example directory first on ``sys.path``, then restores the previous state
on exit so unrelated tests keep their own ``integrations`` package.
"""
saved = {
name: mod
for name, mod in list(sys.modules.items())
if name == "integrations" or name.startswith("integrations.")
}
for name in saved:
del sys.modules[name]
sys.path.insert(0, str(_ROOT))
try:
yield importlib.import_module
finally:
with contextlib.suppress(ValueError):
sys.path.remove(str(_ROOT))
for name in [
name
for name in list(sys.modules)
if name == "integrations" or name.startswith("integrations.")
]:
del sys.modules[name]
sys.modules.update(saved)
@pytest.fixture(autouse=True)
def voice_env(monkeypatch, tmp_path):
monkeypatch.setenv("VOICE_API_KEY", "test-voice-key")
monkeypatch.setenv("VOICE_API_BASE", "https://voice-provider.example.test")
monkeypatch.setenv("VOICE_SERVER_URL_SECRET", "secret-123")
monkeypatch.setenv("PUBLIC_API_BASE_URL", "https://voice-dev.example.test")
monkeypatch.setenv("PRAISONAI_VOICE_DIR", str(tmp_path))
def test_execute_tool_get_current_time():
with _voice_modules() as imp:
tools = imp("integrations.voice.tools")
raw = tools.execute_tool("get_current_time", {})
assert "utc" in raw
def test_tool_calls_webhook_response():
with _voice_modules() as imp:
processor = imp("integrations.voice.processor")
config = imp("integrations.voice.config")
settings = config.load_voice_settings()
payload = {
"message": {
"type": "tool-calls",
"call": {"id": "call-1", "customer": {"number": "+15551234567"}},
"toolCallList": [
{"id": "tc-1", "name": "echo_message", "parameters": {"message": "hello"}}
],
}
}
result = processor.process_voice_webhook("tool-calls", payload, settings=settings)
assert result["results"][0]["name"] == "echo_message"
assert "hello" in result["results"][0]["result"]
def test_assistant_request_returns_default_assistant(monkeypatch):
monkeypatch.setenv("VOICE_ASSISTANT_ID", "asst-123")
with _voice_modules() as imp:
processor = imp("integrations.voice.processor")
config = imp("integrations.voice.config")
settings = config.load_voice_settings()
result = processor.process_voice_webhook(
"assistant-request",
{"message": {"type": "assistant-request", "call": {"id": "c1"}}},
settings=settings,
)
assert result == {"assistantId": "asst-123"}
def test_transcript_appends_final_lines():
with _voice_modules() as imp:
processor = imp("integrations.voice.processor")
config = imp("integrations.voice.config")
store = imp("integrations.voice.store")
settings = config.load_voice_settings()
payload = {
"message": {
"type": "transcript",
"call": {"id": "call-t1"},
"role": "user",
"transcriptType": "final",
"transcript": "I need the time",
}
}
processor.process_voice_webhook("transcript", payload, settings=settings)
record = store.VoiceCallStore.get_call("call-t1")
assert "user: I need the time" in record["transcript"]
def test_transcript_preserves_live_status():
"""A transcript/conversation event must not reset an in-progress call to 'unknown'."""
with _voice_modules() as imp:
processor = imp("integrations.voice.processor")
config = imp("integrations.voice.config")
store = imp("integrations.voice.store")
settings = config.load_voice_settings()
processor.process_voice_webhook(
"status-update",
{"message": {"type": "status-update", "call": {"id": "call-s1"}, "status": "in-progress"}},
settings=settings,
)
processor.process_voice_webhook(
"transcript",
{
"message": {
"type": "transcript",
"call": {"id": "call-s1"},
"role": "user",
"transcriptType": "final",
"transcript": "hello there",
}
},
settings=settings,
)
record = store.VoiceCallStore.get_call("call-s1")
assert record["status"] == "in-progress"
assert "user: hello there" in record["transcript"]
def test_status_updates_are_not_collapsed_by_idempotency_key():
"""Distinct status transitions must each be applied, not dropped as duplicates.
A large static field (``call``) sorts before ``status``; a truncated JSON key
would collapse successive updates to the same prefix and freeze the dashboard
status on the first value.
"""
with _voice_modules() as imp:
processor = imp("integrations.voice.processor")
config = imp("integrations.voice.config")
store = imp("integrations.voice.store")
settings = config.load_voice_settings()
big_call = {
"id": "call-cascade",
"orgId": "org-" * 40,
"assistantId": "asst-" * 40,
"customer": {"number": "+15551234567"},
}
def status_event(status):
return {"message": {"call": big_call, "status": status, "type": "status-update"}}
for status in ("ringing", "in-progress", "ended"):
processor.process_voice_webhook("status-update", status_event(status), settings=settings)
record = store.VoiceCallStore.get_call("call-cascade")
assert record["status"] == "ended"
def test_verify_secret_header():
with _voice_modules() as imp:
verify = imp("integrations.voice.verify")
verify.verify_webhook_request(
secret="secret-123",
headers={"x-voice-webhook-secret": "secret-123"},
)
with pytest.raises(verify.VerificationError):
verify.verify_webhook_request(
secret="secret-123", headers={"x-voice-webhook-secret": "wrong"}
)
def test_create_call_request_body(monkeypatch):
monkeypatch.setenv("VOICE_ASSISTANT_ID", "asst-1")
monkeypatch.setenv("VOICE_PHONE_NUMBER_ID", "pn-1")
with _voice_modules() as imp:
client_mod = imp("integrations.voice.client")
config_mod = imp("integrations.voice.config")
settings = config_mod.load_voice_settings()
client = client_mod.VoiceClient(settings)
client._request = MagicMock(return_value={"id": "call-1"}) # noqa: SLF001
client.create_call(customer_number="+15551234567")
body = client._request.call_args.kwargs["json_body"] # noqa: SLF001
assert body["assistantId"] == "asst-1"
assert body["customer"]["number"] == "+15551234567"
Generated
+1 -1
View File
@@ -207,7 +207,7 @@ wheels = [
[[package]]
name = "aiui"
version = "0.3.142"
version = "0.3.143"
source = { editable = "." }
dependencies = [
{ name = "pydantic" },