diff --git a/docs/usage/integrations/external-task-trigger.md b/docs/usage/integrations/external-task-trigger.md
new file mode 100644
index 000000000..cefd1cf40
--- /dev/null
+++ b/docs/usage/integrations/external-task-trigger.md
@@ -0,0 +1,176 @@
+# External Task Trigger Integration
+
+This document describes how an external backend system (CRM, ticket system, workflow engine, monitoring alert, etc.) can submit structured jobs into AgentTeams through the existing Manager-centered Matrix workflow.
+
+## Why
+
+AgentTeams currently routes all tasks through Matrix conversation — either from a human admin typing in Element Web or from the `scripts/replay-task.sh` CLI bridge. Enterprise systems produce structured jobs, not free-form chat messages. To bridge this gap, an integration needs to:
+
+1. Accept a structured task envelope
+2. Generate traceable task and correlation IDs
+3. Convert the envelope into a Manager-understandable message
+4. Submit through the existing Matrix task submission path
+5. Return a traceable result
+
+## What This Example Demonstrates
+
+```
+External system (CRM, ticket, workflow engine)
+        │
+        ▼
+  Structured JSON task
+  {team, skill, params, metadata}
+        │
+        ▼
+  Task envelope generator
+  (task_id + trace_id)
+        │
+        ▼
+  Manager message builder
+  (deterministic, parseable)
+        │
+        ▼
+  replay-task.sh (Matrix bridge)
+        │
+        ▼
+  AgentTeams Manager
+  (existing collaboration mechanism)
+        │
+        ▼
+  Structured result
+  {task_id, trace_id, status, result}
+```
+
+## How to Run
+
+### Dry-run mode (no Matrix or LLM required)
+
+```bash
+python3 scripts/external-task-trigger.py --task scripts/example-task.json --dry-run
+```
+
+This validates the JSON input, generates task_id and trace_id, builds the Manager message, and returns a simulated result without calling Matrix.
+
+### Live mode (requires running AgentTeams environment)
+
+```bash
+python3 scripts/external-task-trigger.py --task scripts/example-task.json
+```
+
+This uses `scripts/replay-task.sh` to authenticate via Matrix, find or create the DM room with the Manager, send the task, and wait for the Manager's reply.
+
+### Pipe mode
+
+```bash
+cat scripts/example-task.json | python3 scripts/external-task-trigger.py --stdin --dry-run
+```
+
+## Input Schema
+
+Minimum required fields:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `team` | string | Target team name |
+| `skill` | string | Skill or capability name |
+| `params` | object | Skill parameters (JSON object) |
+
+Optional fields:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `metadata.external_job_id` | string | External system's job ID for correlation |
+| `metadata.context` | string | Additional context for the Manager |
+
+Example:
+
+```json
+{
+  "team": "demo-team",
+  "skill": "analyze_request",
+  "params": {
+    "request": "Analyze this customer request and return a recommendation",
+    "priority": "medium"
+  },
+  "metadata": {
+    "external_job_id": "crm-2026-001",
+    "context": "Customer CRM ticket #4821"
+  }
+}
+```
+
+## Output Schema
+
+```json
+{
+  "task_id": "task-da6a32f06eec",
+  "trace_id": "trace-530278c30bf9",
+  "status": "completed",
+  "result": "Manager's response text...",
+  "external_job_id": "crm-2026-001",
+  "submitted_at": "2026-08-11T20:49:58.568769+00:00"
+}
+```
+
+| Field | Description |
+|-------|-------------|
+| `task_id` | Generated by the script; unique per submission |
+| `trace_id` | Generated by the script; unique per submission |
+| `status` | `submitted` (fire-and-forget), `completed`, or `error` |
+| `result` | Manager's reply text (null for fire-and-forget) |
+| `external_job_id` | Passed through from metadata for client-side correlation |
+| `submitted_at` | ISO 8601 UTC timestamp |
+
+## How It Bridges to AgentTeams
+
+The adapter script converts the structured task into a deterministic, parseable message for the Manager:
+
+```
+[EXTERNAL_TASK]
+task_id: task-da6a32f06eec
+trace_id: trace-530278c30bf9
+external_job_id: crm-2026-001
+team: demo-team
+skill: analyze_request
+---
+params: {
+  "request": "Analyze this customer request and return a recommendation",
+  "priority": "medium"
+}
+context: Customer CRM ticket #4821
+```
+
+This message is sent through `scripts/replay-task.sh`, which authenticates as the admin user, finds or creates the DM room with the Manager, sends the message, and waits for the Manager's reply. The Manager receives this as a normal Matrix message and processes it through its existing task coordination workflow.
+
+## Testing
+
+```bash
+# Run deterministic tests (no Matrix or LLM needed)
+python3 -m pytest tests/test_external_task_trigger.py -v
+
+# Or with unittest
+python3 -m unittest tests.test_external_task_trigger -v
+```
+
+Tests cover:
+- Valid JSON task creates task_id + trace_id
+- external_job_id preserved through the pipeline
+- Missing required fields return useful errors
+- Dry-run mode does not require Matrix or LLM
+- Generated Manager message contains trace_id and task_id for log correlation
+- Shell syntax and Python compilation checks pass
+
+## Limitations
+
+**This example is not a production task API.** It demonstrates an integration pattern. Production deployments need:
+
+- **Authentication** — API keys, OAuth, or service-to-service auth between the external system and AgentTeams
+- **Authorization / RBAC** — Control which external systems can submit to which teams
+- **Persistent task state** — Database or state store for task lifecycle tracking
+- **Idempotency** — Prevent duplicate task submissions for the same external_job_id
+- **Retry** — Handle transient Matrix or Manager failures
+- **Timeouts** — Enforce maximum task execution time
+- **Audit retention** — Long-term storage of task submissions and results
+- **Rate limiting** — Protect the Manager from excessive submissions
+
+For production use, consider wrapping this pattern in a service with the above controls, or contributing a formal `POST /tasks` API design to AgentTeams core.
\ No newline at end of file
diff --git a/scripts/example-task.json b/scripts/example-task.json
new file mode 100644
index 000000000..df4e0f051
--- /dev/null
+++ b/scripts/example-task.json
@@ -0,0 +1,12 @@
+{
+  "team": "demo-team",
+  "skill": "analyze_request",
+  "params": {
+    "request": "Analyze this customer request and return a recommendation",
+    "priority": "medium"
+  },
+  "metadata": {
+    "external_job_id": "crm-2026-001",
+    "context": "Customer CRM ticket #4821: Request for quarterly report analysis"
+  }
+}
\ No newline at end of file
diff --git a/scripts/external-task-trigger.py b/scripts/external-task-trigger.py
new file mode 100755
index 000000000..76b3cff9f
--- /dev/null
+++ b/scripts/external-task-trigger.py
@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""External Task Trigger - Bridge structured external jobs into AgentTeams.
+
+This is an integration example, not a stable AgentTeams REST API.
+It demonstrates how an external backend can bridge structured jobs
+into the current Manager-centered Matrix workflow.
+
+Usage:
+    python3 scripts/external-task-trigger.py --task example-task.json
+    python3 scripts/external-task-trigger.py --task example-task.json --dry-run
+    cat example-task.json | python3 scripts/external-task-trigger.py --stdin
+"""
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+REQUIRED_FIELDS = ["team", "skill", "params"]
+
+
+def generate_task_id():
+    return f"task-{uuid.uuid4().hex[:12]}"
+
+
+def generate_trace_id():
+    return f"trace-{uuid.uuid4().hex[:12]}"
+
+
+def load_task(input_file=None, use_stdin=False):
+    if use_stdin:
+        raw = sys.stdin.read()
+        if not raw.strip():
+            raise ValueError("No input received from stdin")
+        return json.loads(raw)
+
+    if input_file:
+        path = Path(input_file)
+        if not path.exists():
+            raise FileNotFoundError(f"Task file not found: {input_file}")
+        with open(path) as f:
+            return json.load(f)
+
+    raise ValueError("Provide --task FILE or --stdin")
+
+
+def validate_task(task):
+    missing = [f for f in REQUIRED_FIELDS if f not in task]
+    if missing:
+        raise ValueError(f"Missing required field(s): {', '.join(missing)}")
+    if not isinstance(task.get("params"), dict):
+        raise ValueError("'params' must be a JSON object")
+    return True
+
+
+def build_manager_message(task, task_id, trace_id):
+    metadata = task.get("metadata", {})
+    external_job_id = metadata.get("external_job_id", "unknown")
+
+    header = (
+        f"[EXTERNAL_TASK]\n"
+        f"task_id: {task_id}\n"
+        f"trace_id: {trace_id}\n"
+        f"external_job_id: {external_job_id}\n"
+        f"team: {task['team']}\n"
+        f"skill: {task['skill']}\n"
+        f"---"
+    )
+
+    params_section = f"params: {json.dumps(task['params'], indent=2)}"
+
+    context = metadata.get("context", "")
+    context_section = f"\ncontext: {context}" if context else ""
+
+    return f"{header}\n{params_section}{context_section}"
+
+
+def run_replay(message, dry_run=False, no_wait=False):
+    if dry_run:
+        return json.dumps({
+            "status": "completed",
+            "result": f"[DRY RUN] Would send task to Manager: {message[:100]}...",
+        })
+
+    project_root = Path(__file__).resolve().parent.parent
+    replay_script = project_root / "scripts" / "replay-task.sh"
+
+    if not replay_script.exists():
+        raise FileNotFoundError(f"replay-task.sh not found at {replay_script}")
+
+    env = os.environ.copy()
+    if no_wait:
+        env["REPLAY_WAIT"] = "0"
+
+    try:
+        result = subprocess.run(
+            [str(replay_script), message],
+            capture_output=True,
+            text=True,
+            timeout=600,
+            env=env,
+            cwd=str(project_root),
+        )
+        if result.returncode != 0:
+            error_msg = result.stderr.strip() or result.stdout.strip()
+            raise RuntimeError(f"replay-task.sh failed: {error_msg}")
+        return result.stdout.strip()
+    except subprocess.TimeoutExpired:
+        raise RuntimeError("replay-task.sh timed out after 600s")
+
+
+def run_task(task_file=None, use_stdin=False, dry_run=False, no_wait=False):
+    try:
+        task = load_task(task_file, use_stdin)
+        validate_task(task)
+    except (ValueError, FileNotFoundError, json.JSONDecodeError) as e:
+        return {
+            "status": "error",
+            "error": str(e),
+        }
+
+    task_id = generate_task_id()
+    trace_id = generate_trace_id()
+    submitted_at = datetime.now(timezone.utc).isoformat()
+
+    manager_message = build_manager_message(task, task_id, trace_id)
+
+    try:
+        raw_result = run_replay(manager_message, dry_run=dry_run, no_wait=no_wait)
+    except Exception as e:
+        return {
+            "task_id": task_id,
+            "trace_id": trace_id,
+            "status": "error",
+            "error": str(e),
+            "external_job_id": task.get("metadata", {}).get("external_job_id"),
+            "submitted_at": submitted_at,
+        }
+
+    if no_wait:
+        status = "submitted"
+        result_text = None
+    elif dry_run:
+        status = "completed"
+        result_text = raw_result
+    else:
+        status = "completed"
+        result_text = raw_result
+
+    return {
+        "task_id": task_id,
+        "trace_id": trace_id,
+        "status": status,
+        "result": result_text,
+        "external_job_id": task.get("metadata", {}).get("external_job_id"),
+        "submitted_at": submitted_at,
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="External Task Trigger - Bridge structured jobs into AgentTeams",
+    )
+    input_group = parser.add_mutually_exclusive_group()
+    input_group.add_argument(
+        "--task",
+        metavar="FILE",
+        help="Path to JSON task file",
+    )
+    input_group.add_argument(
+        "--stdin",
+        action="store_true",
+        help="Read task JSON from stdin",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Simulate submission without calling Matrix",
+    )
+    parser.add_argument(
+        "--no-wait",
+        action="store_true",
+        help="Submit without waiting for Manager reply",
+    )
+
+    args = parser.parse_args()
+
+    if not args.task and not args.stdin:
+        parser.print_help()
+        print("\nError: Provide --task FILE or --stdin", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        output = run_task(
+            task_file=args.task,
+            use_stdin=args.stdin,
+            dry_run=args.dry_run,
+            no_wait=args.no_wait,
+        )
+    except (ValueError, FileNotFoundError) as e:
+        output = {
+            "status": "error",
+            "error": str(e),
+        }
+        print(json.dumps(output, indent=2))
+        sys.exit(1)
+
+    print(json.dumps(output, indent=2))
+
+    if output.get("status") == "error":
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file
diff --git a/tests/test_external_task_trigger.py b/tests/test_external_task_trigger.py
new file mode 100644
index 000000000..7e681c89a
--- /dev/null
+++ b/tests/test_external_task_trigger.py
@@ -0,0 +1,322 @@
+#!/usr/bin/env python3
+"""Deterministic tests for external-task-trigger.py.
+
+Run: python3 -m pytest tests/test_external_task_trigger.py -v
+Or:  python3 -m unittest tests.test_external_task_trigger -v
+Or:  python3 tests/test_external_task_trigger.py
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from io import StringIO
+from pathlib import Path
+from unittest.mock import patch
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent.parent / "scripts"
+sys.path.insert(0, str(SCRIPT_DIR))
+
+import importlib.util
+spec = importlib.util.spec_from_file_location(
+    "external_task_trigger",
+    SCRIPT_DIR / "external-task-trigger.py",
+)
+ext = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(ext)
+
+
+class TestTaskValidation(unittest.TestCase):
+    def test_valid_task(self):
+        task = {
+            "team": "demo-team",
+            "skill": "analyze",
+            "params": {"request": "test"},
+        }
+        result = ext.validate_task(task)
+        self.assertTrue(result)
+
+    def test_missing_team(self):
+        task = {"skill": "analyze", "params": {}}
+        with self.assertRaises(ValueError) as ctx:
+            ext.validate_task(task)
+        self.assertIn("team", str(ctx.exception))
+
+    def test_missing_skill(self):
+        task = {"team": "demo", "params": {}}
+        with self.assertRaises(ValueError) as ctx:
+            ext.validate_task(task)
+        self.assertIn("skill", str(ctx.exception))
+
+    def test_missing_params(self):
+        task = {"team": "demo", "skill": "analyze"}
+        with self.assertRaises(ValueError) as ctx:
+            ext.validate_task(task)
+        self.assertIn("params", str(ctx.exception))
+
+    def test_params_not_dict(self):
+        task = {"team": "demo", "skill": "analyze", "params": "not-a-dict"}
+        with self.assertRaises(ValueError) as ctx:
+            ext.validate_task(task)
+        self.assertIn("params", str(ctx.exception).lower())
+
+    def test_multiple_missing(self):
+        task = {}
+        with self.assertRaises(ValueError) as ctx:
+            ext.validate_task(task)
+        msg = str(ctx.exception)
+        self.assertIn("team", msg)
+        self.assertIn("skill", msg)
+        self.assertIn("params", msg)
+
+
+class TestIDGeneration(unittest.TestCase):
+    def test_task_id_format(self):
+        tid = ext.generate_task_id()
+        self.assertTrue(tid.startswith("task-"))
+        self.assertEqual(len(tid), len("task-") + 12)
+
+    def test_trace_id_format(self):
+        tid = ext.generate_trace_id()
+        self.assertTrue(tid.startswith("trace-"))
+        self.assertEqual(len(tid), len("trace-") + 12)
+
+    def test_ids_are_unique(self):
+        ids = {ext.generate_task_id() for _ in range(100)}
+        self.assertEqual(len(ids), 100)
+
+
+class TestManagerMessage(unittest.TestCase):
+    def test_message_contains_ids(self):
+        task = {
+            "team": "demo-team",
+            "skill": "analyze_request",
+            "params": {"request": "test"},
+            "metadata": {"external_job_id": "crm-001"},
+        }
+        msg = ext.build_manager_message(task, "task-abc", "trace-xyz")
+        self.assertIn("task-abc", msg)
+        self.assertIn("trace-xyz", msg)
+        self.assertIn("crm-001", msg)
+        self.assertIn("demo-team", msg)
+        self.assertIn("analyze_request", msg)
+        self.assertIn("EXTERNAL_TASK", msg)
+
+    def test_message_contains_params(self):
+        task = {
+            "team": "t1",
+            "skill": "s1",
+            "params": {"key": "value", "num": 42},
+            "metadata": {"external_job_id": "ext-1"},
+        }
+        msg = ext.build_manager_message(task, "task-1", "trace-1")
+        self.assertIn("key", msg)
+        self.assertIn("value", msg)
+        self.assertIn("42", msg)
+
+    def test_message_without_optional_metadata(self):
+        task = {"team": "t1", "skill": "s1", "params": {"x": 1}}
+        msg = ext.build_manager_message(task, "task-1", "trace-1")
+        self.assertIn("external_job_id: unknown", msg)
+
+    def test_message_with_context(self):
+        task = {
+            "team": "t1",
+            "skill": "s1",
+            "params": {"x": 1},
+            "metadata": {
+                "external_job_id": "ext-1",
+                "context": "Customer is urgent",
+            },
+        }
+        msg = ext.build_manager_message(task, "task-1", "trace-1")
+        self.assertIn("Customer is urgent", msg)
+
+
+class TestTaskLoading(unittest.TestCase):
+    def test_load_from_file(self):
+        task_data = {
+            "team": "test",
+            "skill": "test",
+            "params": {"k": "v"},
+        }
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task_data, f)
+            f.flush()
+            path = f.name
+
+        try:
+            loaded = ext.load_task(input_file=path)
+            self.assertEqual(loaded, task_data)
+        finally:
+            os.unlink(path)
+
+    def test_load_nonexistent_file(self):
+        with self.assertRaises(FileNotFoundError):
+            ext.load_task(input_file="/nonexistent/path.json")
+
+    def test_load_empty_stdin_raises(self):
+        with patch("sys.stdin", StringIO("")):
+            with self.assertRaises(ValueError):
+                ext.load_task(use_stdin=True)
+
+    def test_no_input_raises(self):
+        with self.assertRaises(ValueError):
+            ext.load_task()
+
+
+class TestDryRun(unittest.TestCase):
+    def test_dry_run_returns_completed(self):
+        task = {
+            "team": "demo-team",
+            "skill": "analyze",
+            "params": {"request": "test"},
+            "metadata": {"external_job_id": "crm-001"},
+        }
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task, f)
+            f.flush()
+            path = f.name
+
+        try:
+            result = ext.run_task(task_file=path, dry_run=True)
+            self.assertEqual(result["status"], "completed")
+            self.assertIn("task_id", result)
+            self.assertIn("trace_id", result)
+            self.assertEqual(result["external_job_id"], "crm-001")
+            self.assertIn("DRY RUN", result["result"])
+        finally:
+            os.unlink(path)
+
+    def test_dry_run_missing_field_returns_error(self):
+        task = {"team": "demo"}
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task, f)
+            f.flush()
+            path = f.name
+
+        try:
+            result = ext.run_task(task_file=path, dry_run=True)
+            self.assertEqual(result["status"], "error")
+            self.assertIn("error", result)
+        finally:
+            os.unlink(path)
+
+    def test_dry_run_no_matrix_needed(self):
+        task = {
+            "team": "t",
+            "skill": "s",
+            "params": {},
+        }
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task, f)
+            f.flush()
+            path = f.name
+
+        try:
+            result = ext.run_task(task_file=path, dry_run=True)
+            self.assertEqual(result["status"], "completed")
+            self.assertIsNotNone(result["task_id"])
+            self.assertIsNotNone(result["trace_id"])
+        finally:
+            os.unlink(path)
+
+
+class TestErrorHandling(unittest.TestCase):
+    def test_invalid_json(self):
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            f.write("not valid json {{{")
+            f.flush()
+            path = f.name
+
+        try:
+            with self.assertRaises(json.JSONDecodeError):
+                ext.load_task(input_file=path)
+        finally:
+            os.unlink(path)
+
+    def test_missing_required_field_gives_useful_error(self):
+        task = {"team": "only-team"}
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task, f)
+            f.flush()
+            path = f.name
+
+        try:
+            result = ext.run_task(task_file=path, dry_run=True)
+            self.assertEqual(result["status"], "error")
+            self.assertIn("skill", result["error"].lower())
+            self.assertIn("params", result["error"].lower())
+        finally:
+            os.unlink(path)
+
+
+class TestTraceability(unittest.TestCase):
+    def test_manager_message_contains_trace_ids(self):
+        task = {
+            "team": "t",
+            "skill": "s",
+            "params": {"x": 1},
+            "metadata": {"external_job_id": "ext-42"},
+        }
+        msg = ext.build_manager_message(task, "task-aaa", "trace-bbb")
+        self.assertIn("task_id: task-aaa", msg)
+        self.assertIn("trace_id: trace-bbb", msg)
+
+    def test_result_preserves_external_job_id(self):
+        task = {
+            "team": "t",
+            "skill": "s",
+            "params": {},
+            "metadata": {"external_job_id": "MY-JOB-99"},
+        }
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task, f)
+            f.flush()
+            path = f.name
+
+        try:
+            result = ext.run_task(task_file=path, dry_run=True)
+            self.assertEqual(result["external_job_id"], "MY-JOB-99")
+        finally:
+            os.unlink(path)
+
+    def test_result_has_submitted_at_timestamp(self):
+        task = {
+            "team": "t",
+            "skill": "s",
+            "params": {},
+        }
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(task, f)
+            f.flush()
+            path = f.name
+
+        try:
+            result = ext.run_task(task_file=path, dry_run=True)
+            self.assertIn("submitted_at", result)
+            self.assertIsNotNone(result["submitted_at"])
+        finally:
+            os.unlink(path)
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file