feat(copilot): add dummy agent generator for testing

Add optional dummy agent generator that returns mock responses matching
the expected format from the external Agent Generator service.

Enable via AGENTGENERATOR_USE_DUMMY=true in settings.

Changes:
- Add agentgenerator_use_dummy setting to settings.py (default: false)
- Create dummy.py with mock implementations of all service functions
- Modify service.py to route to dummy when enabled
- Add startup warning when dummy mode is active

This enables local CoPilot testing without needing the external
Agent Generator service running.

autogpt_platform/backend/backend/api/features/chat/model.py

@@ -2,7 +2,7 @@ import asyncio
 import logging
 import uuid
 from datetime import UTC, datetime
-from typing import Any, cast
+from typing import Any
 from weakref import WeakValueDictionary
 
 from openai.types.chat import (
@@ -104,26 +104,6 @@ class ChatSession(BaseModel):
     successful_agent_runs: dict[str, int] = {}
     successful_agent_schedules: dict[str, int] = {}
 
-    def add_tool_call_to_current_turn(self, tool_call: dict) -> None:
-        """Attach a tool_call to the current turn's assistant message.
-
-        Searches backwards for the most recent assistant message (stopping at
-        any user message boundary). If found, appends the tool_call to it.
-        Otherwise creates a new assistant message with the tool_call.
-        """
-        for msg in reversed(self.messages):
-            if msg.role == "user":
-                break
-            if msg.role == "assistant":
-                if not msg.tool_calls:
-                    msg.tool_calls = []
-                msg.tool_calls.append(tool_call)
-                return
-
-        self.messages.append(
-            ChatMessage(role="assistant", content="", tool_calls=[tool_call])
-        )
-
     @staticmethod
     def new(user_id: str) -> "ChatSession":
         return ChatSession(
@@ -192,47 +172,6 @@ class ChatSession(BaseModel):
             successful_agent_schedules=successful_agent_schedules,
         )
 
-    @staticmethod
-    def _merge_consecutive_assistant_messages(
-        messages: list[ChatCompletionMessageParam],
-    ) -> list[ChatCompletionMessageParam]:
-        """Merge consecutive assistant messages into single messages.
-
-        Long-running tool flows can create split assistant messages: one with
-        text content and another with tool_calls. Anthropic's API requires
-        tool_result blocks to reference a tool_use in the immediately preceding
-        assistant message, so these splits cause 400 errors via OpenRouter.
-        """
-        if len(messages) < 2:
-            return messages
-
-        result: list[ChatCompletionMessageParam] = [messages[0]]
-        for msg in messages[1:]:
-            prev = result[-1]
-            if prev.get("role") != "assistant" or msg.get("role") != "assistant":
-                result.append(msg)
-                continue
-
-            prev = cast(ChatCompletionAssistantMessageParam, prev)
-            curr = cast(ChatCompletionAssistantMessageParam, msg)
-
-            curr_content = curr.get("content") or ""
-            if curr_content:
-                prev_content = prev.get("content") or ""
-                prev["content"] = (
-                    f"{prev_content}\n{curr_content}" if prev_content else curr_content
-                )
-
-            curr_tool_calls = curr.get("tool_calls")
-            if curr_tool_calls:
-                prev_tool_calls = prev.get("tool_calls")
-                prev["tool_calls"] = (
-                    list(prev_tool_calls) + list(curr_tool_calls)
-                    if prev_tool_calls
-                    else list(curr_tool_calls)
-                )
-        return result
-
     def to_openai_messages(self) -> list[ChatCompletionMessageParam]:
         messages = []
         for message in self.messages:
@@ -319,7 +258,7 @@ class ChatSession(BaseModel):
                         name=message.name or "",
                     )
                 )
-        return self._merge_consecutive_assistant_messages(messages)
+        return messages
 
 
 async def _get_session_from_cache(session_id: str) -> ChatSession | None:

autogpt_platform/backend/backend/api/features/chat/model_test.py

@@ -1,16 +1,4 @@
-from typing import cast
-
 import pytest
-from openai.types.chat import (
-    ChatCompletionAssistantMessageParam,
-    ChatCompletionMessageParam,
-    ChatCompletionToolMessageParam,
-    ChatCompletionUserMessageParam,
-)
-from openai.types.chat.chat_completion_message_tool_call_param import (
-    ChatCompletionMessageToolCallParam,
-    Function,
-)
 
 from .model import (
     ChatMessage,
@@ -129,205 +117,3 @@ async def test_chatsession_db_storage(setup_test_user, test_user_id):
                 loaded.tool_calls is not None
             ), f"Tool calls missing for {orig.role} message"
             assert len(orig.tool_calls) == len(loaded.tool_calls)
-
-
-# --------------------------------------------------------------------------- #
-#  _merge_consecutive_assistant_messages                                       #
-# --------------------------------------------------------------------------- #
-
-_tc = ChatCompletionMessageToolCallParam(
-    id="tc1", type="function", function=Function(name="do_stuff", arguments="{}")
-)
-_tc2 = ChatCompletionMessageToolCallParam(
-    id="tc2", type="function", function=Function(name="other", arguments="{}")
-)
-
-
-def test_merge_noop_when_no_consecutive_assistants():
-    """Messages without consecutive assistants are returned unchanged."""
-    msgs = [
-        ChatCompletionUserMessageParam(role="user", content="hi"),
-        ChatCompletionAssistantMessageParam(role="assistant", content="hello"),
-        ChatCompletionUserMessageParam(role="user", content="bye"),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)
-    assert len(merged) == 3
-    assert [m["role"] for m in merged] == ["user", "assistant", "user"]
-
-
-def test_merge_splits_text_and_tool_calls():
-    """The exact bug scenario: text-only assistant followed by tool_calls-only assistant."""
-    msgs = [
-        ChatCompletionUserMessageParam(role="user", content="build agent"),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="Let me build that"
-        ),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="", tool_calls=[_tc]
-        ),
-        ChatCompletionToolMessageParam(role="tool", content="ok", tool_call_id="tc1"),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)
-
-    assert len(merged) == 3
-    assert merged[0]["role"] == "user"
-    assert merged[2]["role"] == "tool"
-    a = cast(ChatCompletionAssistantMessageParam, merged[1])
-    assert a["role"] == "assistant"
-    assert a.get("content") == "Let me build that"
-    assert a.get("tool_calls") == [_tc]
-
-
-def test_merge_combines_tool_calls_from_both():
-    """Both consecutive assistants have tool_calls — they get merged."""
-    msgs: list[ChatCompletionAssistantMessageParam] = [
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="text", tool_calls=[_tc]
-        ),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="", tool_calls=[_tc2]
-        ),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)  # type: ignore[arg-type]
-
-    assert len(merged) == 1
-    a = cast(ChatCompletionAssistantMessageParam, merged[0])
-    assert a.get("tool_calls") == [_tc, _tc2]
-    assert a.get("content") == "text"
-
-
-def test_merge_three_consecutive_assistants():
-    """Three consecutive assistants collapse into one."""
-    msgs: list[ChatCompletionAssistantMessageParam] = [
-        ChatCompletionAssistantMessageParam(role="assistant", content="a"),
-        ChatCompletionAssistantMessageParam(role="assistant", content="b"),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="", tool_calls=[_tc]
-        ),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)  # type: ignore[arg-type]
-
-    assert len(merged) == 1
-    a = cast(ChatCompletionAssistantMessageParam, merged[0])
-    assert a.get("content") == "a\nb"
-    assert a.get("tool_calls") == [_tc]
-
-
-def test_merge_empty_and_single_message():
-    """Edge cases: empty list and single message."""
-    assert ChatSession._merge_consecutive_assistant_messages([]) == []
-
-    single: list[ChatCompletionMessageParam] = [
-        ChatCompletionUserMessageParam(role="user", content="hi")
-    ]
-    assert ChatSession._merge_consecutive_assistant_messages(single) == single
-
-
-# --------------------------------------------------------------------------- #
-#  add_tool_call_to_current_turn                                               #
-# --------------------------------------------------------------------------- #
-
-_raw_tc = {
-    "id": "tc1",
-    "type": "function",
-    "function": {"name": "f", "arguments": "{}"},
-}
-_raw_tc2 = {
-    "id": "tc2",
-    "type": "function",
-    "function": {"name": "g", "arguments": "{}"},
-}
-
-
-def test_add_tool_call_appends_to_existing_assistant():
-    """When the last assistant is from the current turn, tool_call is added to it."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="hi"),
-        ChatMessage(role="assistant", content="working on it"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-
-    assert len(session.messages) == 2  # no new message created
-    assert session.messages[1].tool_calls == [_raw_tc]
-
-
-def test_add_tool_call_creates_assistant_when_none_exists():
-    """When there's no current-turn assistant, a new one is created."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="hi"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-
-    assert len(session.messages) == 2
-    assert session.messages[1].role == "assistant"
-    assert session.messages[1].tool_calls == [_raw_tc]
-
-
-def test_add_tool_call_does_not_cross_user_boundary():
-    """A user message acts as a boundary — previous assistant is not modified."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="assistant", content="old turn"),
-        ChatMessage(role="user", content="new message"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-
-    assert len(session.messages) == 3  # new assistant was created
-    assert session.messages[0].tool_calls is None  # old assistant untouched
-    assert session.messages[2].role == "assistant"
-    assert session.messages[2].tool_calls == [_raw_tc]
-
-
-def test_add_tool_call_multiple_times():
-    """Multiple long-running tool calls accumulate on the same assistant."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="hi"),
-        ChatMessage(role="assistant", content="doing stuff"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-    # Simulate a pending tool result in between (like _yield_tool_call does)
-    session.messages.append(
-        ChatMessage(role="tool", content="pending", tool_call_id="tc1")
-    )
-    session.add_tool_call_to_current_turn(_raw_tc2)
-
-    assert len(session.messages) == 3  # user, assistant, tool — no extra assistant
-    assert session.messages[1].tool_calls == [_raw_tc, _raw_tc2]
-
-
-def test_to_openai_messages_merges_split_assistants():
-    """End-to-end: session with split assistants produces valid OpenAI messages."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="build agent"),
-        ChatMessage(role="assistant", content="Let me build that"),
-        ChatMessage(
-            role="assistant",
-            content="",
-            tool_calls=[
-                {
-                    "id": "tc1",
-                    "type": "function",
-                    "function": {"name": "create_agent", "arguments": "{}"},
-                }
-            ],
-        ),
-        ChatMessage(role="tool", content="done", tool_call_id="tc1"),
-        ChatMessage(role="assistant", content="Saved!"),
-        ChatMessage(role="user", content="show me an example run"),
-    ]
-    openai_msgs = session.to_openai_messages()
-
-    # The two consecutive assistants at index 1,2 should be merged
-    roles = [m["role"] for m in openai_msgs]
-    assert roles == ["user", "assistant", "tool", "assistant", "user"]
-
-    # The merged assistant should have both content and tool_calls
-    merged = cast(ChatCompletionAssistantMessageParam, openai_msgs[1])
-    assert merged.get("content") == "Let me build that"
-    tc_list = merged.get("tool_calls")
-    assert tc_list is not None and len(list(tc_list)) == 1
-    assert list(tc_list)[0]["id"] == "tc1"

autogpt_platform/backend/backend/api/features/chat/service.py

@@ -800,13 +800,9 @@ async def stream_chat_completion(
         # Build the messages list in the correct order
         messages_to_save: list[ChatMessage] = []
 
-        # Add assistant message with tool_calls if any.
-        # Use extend (not assign) to preserve tool_calls already added by
-        # _yield_tool_call for long-running tools.
+        # Add assistant message with tool_calls if any
         if accumulated_tool_calls:
-            if not assistant_response.tool_calls:
-                assistant_response.tool_calls = []
-            assistant_response.tool_calls.extend(accumulated_tool_calls)
+            assistant_response.tool_calls = accumulated_tool_calls
             logger.info(
                 f"Added {len(accumulated_tool_calls)} tool calls to assistant message"
             )
@@ -1408,9 +1404,13 @@ async def _yield_tool_call(
                 operation_id=operation_id,
             )
 
-            # Attach the tool_call to the current turn's assistant message
-            # (or create one if this is a tool-only response with no text).
-            session.add_tool_call_to_current_turn(tool_calls[yield_idx])
+            # Save assistant message with tool_call FIRST (required by LLM)
+            assistant_message = ChatMessage(
+                role="assistant",
+                content="",
+                tool_calls=[tool_calls[yield_idx]],
+            )
+            session.messages.append(assistant_message)
 
             # Then save pending tool result
             pending_message = ChatMessage(

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/dummy.py

@@ -0,0 +1,152 @@
+"""Dummy Agent Generator for testing.
+
+Returns mock responses matching the format expected from the external service.
+Enable via AGENTGENERATOR_USE_DUMMY=true in settings.
+
+WARNING: This is for testing only. Do not use in production.
+"""
+
+import logging
+import uuid
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Dummy decomposition result (instructions type)
+DUMMY_DECOMPOSITION_RESULT: dict[str, Any] = {
+    "type": "instructions",
+    "steps": [
+        {
+            "description": "Get input from user",
+            "action": "input",
+            "block_name": "AgentInputBlock",
+        },
+        {
+            "description": "Process the input",
+            "action": "process",
+            "block_name": "TextFormatterBlock",
+        },
+        {
+            "description": "Return output to user",
+            "action": "output",
+            "block_name": "AgentOutputBlock",
+        },
+    ],
+}
+
+# Block IDs from backend/blocks/io.py
+AGENT_INPUT_BLOCK_ID = "c0a8e994-ebf1-4a9c-a4d8-89d09c86741b"
+AGENT_OUTPUT_BLOCK_ID = "363ae599-353e-4804-937e-b2ee3cef3da4"
+
+
+def _generate_dummy_agent_json() -> dict[str, Any]:
+    """Generate a minimal valid agent JSON for testing."""
+    input_node_id = str(uuid.uuid4())
+    output_node_id = str(uuid.uuid4())
+
+    return {
+        "id": str(uuid.uuid4()),
+        "version": 1,
+        "is_active": True,
+        "name": "Dummy Test Agent",
+        "description": "A dummy agent generated for testing purposes",
+        "nodes": [
+            {
+                "id": input_node_id,
+                "block_id": AGENT_INPUT_BLOCK_ID,
+                "input_default": {
+                    "name": "input",
+                    "title": "Input",
+                    "description": "Enter your input",
+                    "placeholder_values": [],
+                },
+                "metadata": {"position": {"x": 0, "y": 0}},
+            },
+            {
+                "id": output_node_id,
+                "block_id": AGENT_OUTPUT_BLOCK_ID,
+                "input_default": {
+                    "name": "output",
+                    "title": "Output",
+                    "description": "Agent output",
+                    "format": "{output}",
+                },
+                "metadata": {"position": {"x": 400, "y": 0}},
+            },
+        ],
+        "links": [
+            {
+                "id": str(uuid.uuid4()),
+                "source_id": input_node_id,
+                "sink_id": output_node_id,
+                "source_name": "result",
+                "sink_name": "value",
+                "is_static": False,
+            },
+        ],
+    }
+
+
+async def decompose_goal_dummy(
+    description: str,
+    context: str = "",
+    library_agents: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Return dummy decomposition result."""
+    logger.info("Using dummy agent generator for decompose_goal")
+    return DUMMY_DECOMPOSITION_RESULT.copy()
+
+
+async def generate_agent_dummy(
+    instructions: dict[str, Any],
+    library_agents: list[dict[str, Any]] | None = None,
+    operation_id: str | None = None,
+    task_id: str | None = None,
+) -> dict[str, Any]:
+    """Return dummy agent JSON."""
+    logger.info("Using dummy agent generator for generate_agent")
+    return _generate_dummy_agent_json()
+
+
+async def generate_agent_patch_dummy(
+    update_request: str,
+    current_agent: dict[str, Any],
+    library_agents: list[dict[str, Any]] | None = None,
+    operation_id: str | None = None,
+    task_id: str | None = None,
+) -> dict[str, Any]:
+    """Return dummy patched agent (returns the current agent with updated description)."""
+    logger.info("Using dummy agent generator for generate_agent_patch")
+    patched = current_agent.copy()
+    patched["description"] = (
+        f"{current_agent.get('description', '')} (updated: {update_request})"
+    )
+    return patched
+
+
+async def customize_template_dummy(
+    template_agent: dict[str, Any],
+    modification_request: str,
+    context: str = "",
+) -> dict[str, Any]:
+    """Return dummy customized template (returns template with updated description)."""
+    logger.info("Using dummy agent generator for customize_template")
+    customized = template_agent.copy()
+    customized["description"] = (
+        f"{template_agent.get('description', '')} (customized: {modification_request})"
+    )
+    return customized
+
+
+async def get_blocks_dummy() -> list[dict[str, Any]]:
+    """Return dummy blocks list."""
+    logger.info("Using dummy agent generator for get_blocks")
+    return [
+        {"id": AGENT_INPUT_BLOCK_ID, "name": "AgentInputBlock"},
+        {"id": AGENT_OUTPUT_BLOCK_ID, "name": "AgentOutputBlock"},
+    ]
+
+
+async def health_check_dummy() -> bool:
+    """Always returns healthy for dummy service."""
+    return True

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/service.py

@@ -12,8 +12,19 @@ import httpx
 
 from backend.util.settings import Settings
 
+from .dummy import (
+    customize_template_dummy,
+    decompose_goal_dummy,
+    generate_agent_dummy,
+    generate_agent_patch_dummy,
+    get_blocks_dummy,
+    health_check_dummy,
+)
+
 logger = logging.getLogger(__name__)
 
+_dummy_mode_warned = False
+
 
 def _create_error_response(
     error_message: str,
@@ -90,10 +101,26 @@ def _get_settings() -> Settings:
     return _settings
 
 
-def is_external_service_configured() -> bool:
-    """Check if external Agent Generator service is configured."""
+def _is_dummy_mode() -> bool:
+    """Check if dummy mode is enabled for testing."""
+    global _dummy_mode_warned
     settings = _get_settings()
-    return bool(settings.config.agentgenerator_host)
+    is_dummy = bool(settings.config.agentgenerator_use_dummy)
+    if is_dummy and not _dummy_mode_warned:
+        logger.warning(
+            "Agent Generator running in DUMMY MODE - returning mock responses. "
+            "Do not use in production!"
+        )
+        _dummy_mode_warned = True
+    return is_dummy
+
+
+def is_external_service_configured() -> bool:
+    """Check if external Agent Generator service is configured (or dummy mode)."""
+    settings = _get_settings()
+    return bool(settings.config.agentgenerator_host) or bool(
+        settings.config.agentgenerator_use_dummy
+    )
 
 
 def _get_base_url() -> str:
@@ -137,6 +164,9 @@ async def decompose_goal_external(
         - {"type": "error", "error": "...", "error_type": "..."} on error
         Or None on unexpected error
     """
+    if _is_dummy_mode():
+        return await decompose_goal_dummy(description, context, library_agents)
+
     client = _get_client()
 
     if context:
@@ -226,6 +256,11 @@ async def generate_agent_external(
     Returns:
         Agent JSON dict, {"status": "accepted"} for async, or error dict {"type": "error", ...} on error
     """
+    if _is_dummy_mode():
+        return await generate_agent_dummy(
+            instructions, library_agents, operation_id, task_id
+        )
+
     client = _get_client()
 
     # Build request payload
@@ -297,6 +332,11 @@ async def generate_agent_patch_external(
     Returns:
         Updated agent JSON, clarifying questions dict, {"status": "accepted"} for async, or error dict on error
     """
+    if _is_dummy_mode():
+        return await generate_agent_patch_dummy(
+            update_request, current_agent, library_agents, operation_id, task_id
+        )
+
     client = _get_client()
 
     # Build request payload
@@ -383,6 +423,11 @@ async def customize_template_external(
     Returns:
         Customized agent JSON, clarifying questions dict, or error dict on error
     """
+    if _is_dummy_mode():
+        return await customize_template_dummy(
+            template_agent, modification_request, context
+        )
+
     client = _get_client()
 
     request = modification_request
@@ -445,6 +490,9 @@ async def get_blocks_external() -> list[dict[str, Any]] | None:
     Returns:
         List of block info dicts or None on error
     """
+    if _is_dummy_mode():
+        return await get_blocks_dummy()
+
     client = _get_client()
 
     try:
@@ -478,6 +526,9 @@ async def health_check() -> bool:
     if not is_external_service_configured():
         return False
 
+    if _is_dummy_mode():
+        return await health_check_dummy()
+
     client = _get_client()
 
     try:

autogpt_platform/backend/backend/blocks/telegram/__init__.py

@@ -1 +0,0 @@
-# Telegram Bot blocks for AutoGPT Platform

autogpt_platform/backend/backend/blocks/telegram/_api.py

@@ -1,159 +0,0 @@
-"""
-Telegram Bot API helper functions.
-
-Provides utilities for making authenticated requests to the Telegram Bot API.
-"""
-
-import logging
-from io import BytesIO
-from typing import Any, Optional
-
-from backend.data.model import APIKeyCredentials
-from backend.util.request import Requests
-
-logger = logging.getLogger(__name__)
-
-TELEGRAM_API_BASE = "https://api.telegram.org"
-
-
-class TelegramAPIException(Exception):
-    """Exception raised for Telegram API errors."""
-
-    def __init__(self, message: str, error_code: int = 0):
-        super().__init__(message)
-        self.error_code = error_code
-
-
-def get_bot_api_url(bot_token: str, method: str) -> str:
-    """Construct Telegram Bot API URL for a method."""
-    return f"{TELEGRAM_API_BASE}/bot{bot_token}/{method}"
-
-
-def get_file_url(bot_token: str, file_path: str) -> str:
-    """Construct Telegram file download URL."""
-    return f"{TELEGRAM_API_BASE}/file/bot{bot_token}/{file_path}"
-
-
-async def call_telegram_api(
-    credentials: APIKeyCredentials,
-    method: str,
-    data: Optional[dict[str, Any]] = None,
-) -> dict[str, Any]:
-    """
-    Make a request to the Telegram Bot API.
-
-    Args:
-        credentials: Bot token credentials
-        method: API method name (e.g., "sendMessage", "getFile")
-        data: Request parameters
-
-    Returns:
-        API response result
-
-    Raises:
-        TelegramAPIException: If the API returns an error
-    """
-    token = credentials.api_key.get_secret_value()
-    url = get_bot_api_url(token, method)
-
-    response = await Requests().post(url, json=data or {})
-    result = response.json()
-
-    if not result.get("ok"):
-        error_code = result.get("error_code", 0)
-        description = result.get("description", "Unknown error")
-        raise TelegramAPIException(description, error_code)
-
-    return result.get("result", {})
-
-
-async def call_telegram_api_with_file(
-    credentials: APIKeyCredentials,
-    method: str,
-    file_field: str,
-    file_data: bytes,
-    filename: str,
-    content_type: str,
-    data: Optional[dict[str, Any]] = None,
-) -> dict[str, Any]:
-    """
-    Make a multipart/form-data request to the Telegram Bot API with a file upload.
-
-    Args:
-        credentials: Bot token credentials
-        method: API method name (e.g., "sendPhoto", "sendVoice")
-        file_field: Form field name for the file (e.g., "photo", "voice")
-        file_data: Raw file bytes
-        filename: Filename for the upload
-        content_type: MIME type of the file
-        data: Additional form parameters
-
-    Returns:
-        API response result
-
-    Raises:
-        TelegramAPIException: If the API returns an error
-    """
-    token = credentials.api_key.get_secret_value()
-    url = get_bot_api_url(token, method)
-
-    files = [(file_field, (filename, BytesIO(file_data), content_type))]
-
-    response = await Requests().post(url, files=files, data=data or {})
-    result = response.json()
-
-    if not result.get("ok"):
-        error_code = result.get("error_code", 0)
-        description = result.get("description", "Unknown error")
-        raise TelegramAPIException(description, error_code)
-
-    return result.get("result", {})
-
-
-async def get_file_info(credentials: APIKeyCredentials, file_id: str) -> dict[str, Any]:
-    """
-    Get file information from Telegram.
-
-    Args:
-        credentials: Bot token credentials
-        file_id: Telegram file_id from message
-
-    Returns:
-        File info dict containing file_id, file_unique_id, file_size, file_path
-    """
-    return await call_telegram_api(credentials, "getFile", {"file_id": file_id})
-
-
-async def get_file_download_url(credentials: APIKeyCredentials, file_id: str) -> str:
-    """
-    Get the download URL for a Telegram file.
-
-    Args:
-        credentials: Bot token credentials
-        file_id: Telegram file_id from message
-
-    Returns:
-        Full download URL
-    """
-    token = credentials.api_key.get_secret_value()
-    result = await get_file_info(credentials, file_id)
-    file_path = result.get("file_path")
-    if not file_path:
-        raise TelegramAPIException("No file_path returned from getFile")
-    return get_file_url(token, file_path)
-
-
-async def download_telegram_file(credentials: APIKeyCredentials, file_id: str) -> bytes:
-    """
-    Download a file from Telegram servers.
-
-    Args:
-        credentials: Bot token credentials
-        file_id: Telegram file_id
-
-    Returns:
-        File content as bytes
-    """
-    url = await get_file_download_url(credentials, file_id)
-    response = await Requests().get(url)
-    return response.content

autogpt_platform/backend/backend/blocks/telegram/_auth.py

@@ -1,43 +0,0 @@
-"""
-Telegram Bot credentials handling.
-
-Telegram bots use an API key (bot token) obtained from @BotFather.
-"""
-
-from typing import Literal
-
-from pydantic import SecretStr
-
-from backend.data.model import APIKeyCredentials, CredentialsField, CredentialsMetaInput
-from backend.integrations.providers import ProviderName
-
-# Bot token credentials (API key style)
-TelegramCredentials = APIKeyCredentials
-TelegramCredentialsInput = CredentialsMetaInput[
-    Literal[ProviderName.TELEGRAM], Literal["api_key"]
-]
-
-
-def TelegramCredentialsField() -> TelegramCredentialsInput:
-    """Creates a Telegram bot token credentials field."""
-    return CredentialsField(
-        description="Telegram Bot API token from @BotFather. "
-        "Create a bot at https://t.me/BotFather to get your token."
-    )
-
-
-# Test credentials for unit tests
-TEST_CREDENTIALS = APIKeyCredentials(
-    id="01234567-89ab-cdef-0123-456789abcdef",
-    provider="telegram",
-    api_key=SecretStr("test_telegram_bot_token"),
-    title="Mock Telegram Bot Token",
-    expires_at=None,
-)
-
-TEST_CREDENTIALS_INPUT = {
-    "provider": TEST_CREDENTIALS.provider,
-    "id": TEST_CREDENTIALS.id,
-    "type": TEST_CREDENTIALS.type,
-    "title": TEST_CREDENTIALS.title,
-}

autogpt_platform/backend/backend/blocks/telegram/triggers.py

@@ -1,352 +0,0 @@
-"""
-Telegram trigger blocks for receiving messages via webhooks.
-"""
-
-import logging
-
-from pydantic import BaseModel
-
-from backend.data.block import (
-    Block,
-    BlockCategory,
-    BlockOutput,
-    BlockSchemaInput,
-    BlockSchemaOutput,
-    BlockWebhookConfig,
-)
-from backend.data.model import SchemaField
-from backend.integrations.providers import ProviderName
-from backend.integrations.webhooks.telegram import TelegramWebhookType
-
-from ._auth import (
-    TEST_CREDENTIALS,
-    TEST_CREDENTIALS_INPUT,
-    TelegramCredentialsField,
-    TelegramCredentialsInput,
-)
-
-logger = logging.getLogger(__name__)
-
-
-# Example payload for testing
-EXAMPLE_MESSAGE_PAYLOAD = {
-    "update_id": 123456789,
-    "message": {
-        "message_id": 1,
-        "from": {
-            "id": 12345678,
-            "is_bot": False,
-            "first_name": "John",
-            "last_name": "Doe",
-            "username": "johndoe",
-            "language_code": "en",
-        },
-        "chat": {
-            "id": 12345678,
-            "first_name": "John",
-            "last_name": "Doe",
-            "username": "johndoe",
-            "type": "private",
-        },
-        "date": 1234567890,
-        "text": "Hello, bot!",
-    },
-}
-
-
-class TelegramTriggerBase:
-    """Base class for Telegram trigger blocks."""
-
-    class Input(BlockSchemaInput):
-        credentials: TelegramCredentialsInput = TelegramCredentialsField()
-        payload: dict = SchemaField(hidden=True, default_factory=dict)
-
-
-class TelegramMessageTriggerBlock(TelegramTriggerBase, Block):
-    """
-    Triggers when a message is received by your Telegram bot.
-
-    Supports text, photos, voice messages, and audio files.
-    Connect the outputs to other blocks to process messages and send responses.
-    """
-
-    class Input(TelegramTriggerBase.Input):
-        class EventsFilter(BaseModel):
-            """Filter for message types to receive."""
-
-            text: bool = True
-            photo: bool = False
-            voice: bool = False
-            audio: bool = False
-            document: bool = False
-            video: bool = False
-
-        events: EventsFilter = SchemaField(
-            title="Message Types", description="Types of messages to receive"
-        )
-
-    class Output(BlockSchemaOutput):
-        payload: dict = SchemaField(
-            description="The complete webhook payload from Telegram"
-        )
-        chat_id: int = SchemaField(
-            description="The chat ID where the message was received. "
-            "Use this to send replies."
-        )
-        message_id: int = SchemaField(description="The unique message ID")
-        user_id: int = SchemaField(description="The user ID who sent the message")
-        username: str = SchemaField(description="Username of the sender (may be empty)")
-        first_name: str = SchemaField(description="First name of the sender")
-        event: str = SchemaField(
-            description="The message type (text, photo, voice, audio, etc.)"
-        )
-        text: str = SchemaField(
-            description="Text content of the message (for text messages)"
-        )
-        photo_file_id: str = SchemaField(
-            description="File ID of the photo (for photo messages). "
-            "Use GetTelegramFileBlock to download."
-        )
-        voice_file_id: str = SchemaField(
-            description="File ID of the voice message (for voice messages). "
-            "Use GetTelegramFileBlock to download."
-        )
-        audio_file_id: str = SchemaField(
-            description="File ID of the audio file (for audio messages). "
-            "Use GetTelegramFileBlock to download."
-        )
-        file_id: str = SchemaField(
-            description="File ID for document/video messages. "
-            "Use GetTelegramFileBlock to download."
-        )
-        file_name: str = SchemaField(
-            description="Original filename (for document/audio messages)"
-        )
-        caption: str = SchemaField(description="Caption for media messages")
-
-    def __init__(self):
-        super().__init__(
-            id="4435e4e0-df6e-4301-8f35-ad70b12fc9ec",
-            description="Triggers when a message is received by your Telegram bot. "
-            "Supports text, photos, voice messages, and audio files.",
-            categories={BlockCategory.SOCIAL},
-            input_schema=TelegramMessageTriggerBlock.Input,
-            output_schema=TelegramMessageTriggerBlock.Output,
-            webhook_config=BlockWebhookConfig(
-                provider=ProviderName.TELEGRAM,
-                webhook_type=TelegramWebhookType.BOT,
-                resource_format="bot",
-                event_filter_input="events",
-                event_format="message.{event}",
-            ),
-            test_input={
-                "events": {"text": True, "photo": True},
-                "credentials": TEST_CREDENTIALS_INPUT,
-                "payload": EXAMPLE_MESSAGE_PAYLOAD,
-            },
-            test_credentials=TEST_CREDENTIALS,
-            test_output=[
-                ("payload", EXAMPLE_MESSAGE_PAYLOAD),
-                ("chat_id", 12345678),
-                ("message_id", 1),
-                ("user_id", 12345678),
-                ("username", "johndoe"),
-                ("first_name", "John"),
-                ("event", "text"),
-                ("text", "Hello, bot!"),
-                ("photo_file_id", ""),
-                ("voice_file_id", ""),
-                ("audio_file_id", ""),
-                ("file_id", ""),
-                ("file_name", ""),
-                ("caption", ""),
-            ],
-        )
-
-    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
-        payload = input_data.payload
-        message = payload.get("message", {})
-
-        # Extract common fields
-        chat = message.get("chat", {})
-        sender = message.get("from", {})
-
-        yield "payload", payload
-        yield "chat_id", chat.get("id", 0)
-        yield "message_id", message.get("message_id", 0)
-        yield "user_id", sender.get("id", 0)
-        yield "username", sender.get("username", "")
-        yield "first_name", sender.get("first_name", "")
-
-        # Determine message type and extract content
-        if "text" in message:
-            yield "event", "text"
-            yield "text", message.get("text", "")
-            yield "photo_file_id", ""
-            yield "voice_file_id", ""
-            yield "audio_file_id", ""
-            yield "file_id", ""
-            yield "file_name", ""
-            yield "caption", ""
-        elif "photo" in message:
-            # Get the largest photo (last in array)
-            photos = message.get("photo", [])
-            photo_fid = photos[-1]["file_id"] if photos else ""
-            yield "event", "photo"
-            yield "text", ""
-            yield "photo_file_id", photo_fid
-            yield "voice_file_id", ""
-            yield "audio_file_id", ""
-            yield "file_id", ""
-            yield "file_name", ""
-            yield "caption", message.get("caption", "")
-        elif "voice" in message:
-            voice = message.get("voice", {})
-            yield "event", "voice"
-            yield "text", ""
-            yield "photo_file_id", ""
-            yield "voice_file_id", voice.get("file_id", "")
-            yield "audio_file_id", ""
-            yield "file_id", ""
-            yield "file_name", ""
-            yield "caption", message.get("caption", "")
-        elif "audio" in message:
-            audio = message.get("audio", {})
-            yield "event", "audio"
-            yield "text", ""
-            yield "photo_file_id", ""
-            yield "voice_file_id", ""
-            yield "audio_file_id", audio.get("file_id", "")
-            yield "file_id", ""
-            yield "file_name", audio.get("file_name", "")
-            yield "caption", message.get("caption", "")
-        elif "document" in message:
-            document = message.get("document", {})
-            yield "event", "document"
-            yield "text", ""
-            yield "photo_file_id", ""
-            yield "voice_file_id", ""
-            yield "audio_file_id", ""
-            yield "file_id", document.get("file_id", "")
-            yield "file_name", document.get("file_name", "")
-            yield "caption", message.get("caption", "")
-        elif "video" in message:
-            video = message.get("video", {})
-            yield "event", "video"
-            yield "text", ""
-            yield "photo_file_id", ""
-            yield "voice_file_id", ""
-            yield "audio_file_id", ""
-            yield "file_id", video.get("file_id", "")
-            yield "file_name", video.get("file_name", "")
-            yield "caption", message.get("caption", "")
-        else:
-            yield "event", "other"
-            yield "text", ""
-            yield "photo_file_id", ""
-            yield "voice_file_id", ""
-            yield "audio_file_id", ""
-            yield "file_id", ""
-            yield "file_name", ""
-            yield "caption", ""
-
-
-# Example payload for reaction trigger testing
-EXAMPLE_REACTION_PAYLOAD = {
-    "update_id": 123456790,
-    "message_reaction": {
-        "chat": {
-            "id": 12345678,
-            "first_name": "John",
-            "last_name": "Doe",
-            "username": "johndoe",
-            "type": "private",
-        },
-        "message_id": 42,
-        "user": {
-            "id": 12345678,
-            "is_bot": False,
-            "first_name": "John",
-            "username": "johndoe",
-        },
-        "date": 1234567890,
-        "new_reaction": [{"type": "emoji", "emoji": "👍"}],
-        "old_reaction": [],
-    },
-}
-
-
-class TelegramMessageReactionTriggerBlock(TelegramTriggerBase, Block):
-    """
-    Triggers when a reaction to a message is changed.
-
-    Works automatically in private chats. In group chats, the bot must be
-    an administrator to receive reaction updates.
-    """
-
-    class Input(TelegramTriggerBase.Input):
-        pass
-
-    class Output(BlockSchemaOutput):
-        payload: dict = SchemaField(
-            description="The complete webhook payload from Telegram"
-        )
-        chat_id: int = SchemaField(
-            description="The chat ID where the reaction occurred"
-        )
-        message_id: int = SchemaField(description="The message ID that was reacted to")
-        user_id: int = SchemaField(description="The user ID who changed the reaction")
-        username: str = SchemaField(description="Username of the user (may be empty)")
-        new_reactions: list = SchemaField(
-            description="List of new reactions on the message"
-        )
-        old_reactions: list = SchemaField(
-            description="List of previous reactions on the message"
-        )
-
-    def __init__(self):
-        super().__init__(
-            id="82525328-9368-4966-8f0c-cd78e80181fd",
-            description="Triggers when a reaction to a message is changed. "
-            "Works in private chats automatically. "
-            "In groups, the bot must be an administrator.",
-            categories={BlockCategory.SOCIAL},
-            input_schema=TelegramMessageReactionTriggerBlock.Input,
-            output_schema=TelegramMessageReactionTriggerBlock.Output,
-            webhook_config=BlockWebhookConfig(
-                provider=ProviderName.TELEGRAM,
-                webhook_type=TelegramWebhookType.BOT,
-                resource_format="bot",
-                event_filter_input="",
-                event_format="message_reaction",
-            ),
-            test_input={
-                "credentials": TEST_CREDENTIALS_INPUT,
-                "payload": EXAMPLE_REACTION_PAYLOAD,
-            },
-            test_credentials=TEST_CREDENTIALS,
-            test_output=[
-                ("payload", EXAMPLE_REACTION_PAYLOAD),
-                ("chat_id", 12345678),
-                ("message_id", 42),
-                ("user_id", 12345678),
-                ("username", "johndoe"),
-                ("new_reactions", [{"type": "emoji", "emoji": "👍"}]),
-                ("old_reactions", []),
-            ],
-        )
-
-    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
-        payload = input_data.payload
-        reaction = payload.get("message_reaction", {})
-
-        chat = reaction.get("chat", {})
-        user = reaction.get("user", {})
-
-        yield "payload", payload
-        yield "chat_id", chat.get("id", 0)
-        yield "message_id", reaction.get("message_id", 0)
-        yield "user_id", user.get("id", 0)
-        yield "username", user.get("username", "")
-        yield "new_reactions", reaction.get("new_reaction", [])
-        yield "old_reactions", reaction.get("old_reaction", [])

autogpt_platform/backend/backend/integrations/providers.py

@@ -46,7 +46,6 @@ class ProviderName(str, Enum):
     SLANT3D = "slant3d"
     SMARTLEAD = "smartlead"
     SMTP = "smtp"
-    TELEGRAM = "telegram"
     TWITTER = "twitter"
     TODOIST = "todoist"
     UNREAL_SPEECH = "unreal_speech"

autogpt_platform/backend/backend/integrations/webhooks/__init__.py

@@ -15,7 +15,6 @@ def load_webhook_managers() -> dict["ProviderName", type["BaseWebhooksManager"]]
     from .compass import CompassWebhookManager
     from .github import GithubWebhooksManager
     from .slant3d import Slant3DWebhooksManager
-    from .telegram import TelegramWebhooksManager
 
     webhook_managers.update(
         {
@@ -24,7 +23,6 @@ def load_webhook_managers() -> dict["ProviderName", type["BaseWebhooksManager"]]
                 CompassWebhookManager,
                 GithubWebhooksManager,
                 Slant3DWebhooksManager,
-                TelegramWebhooksManager,
             ]
         }
     )

autogpt_platform/backend/backend/integrations/webhooks/telegram.py

@@ -1,166 +0,0 @@
-"""
-Telegram Bot API Webhooks Manager.
-
-Handles webhook registration and validation for Telegram bots.
-"""
-
-import logging
-
-from fastapi import HTTPException, Request
-from strenum import StrEnum
-
-from backend.data import integrations
-from backend.data.model import APIKeyCredentials, Credentials
-from backend.integrations.providers import ProviderName
-from backend.util.request import Requests
-
-from ._base import BaseWebhooksManager
-
-logger = logging.getLogger(__name__)
-
-
-class TelegramWebhookType(StrEnum):
-    BOT = "bot"
-
-
-class TelegramWebhooksManager(BaseWebhooksManager[TelegramWebhookType]):
-    """
-    Manages Telegram bot webhooks.
-
-    Telegram webhooks are registered via the setWebhook API method.
-    Incoming requests are validated using the secret_token header.
-    """
-
-    PROVIDER_NAME = ProviderName.TELEGRAM
-    WebhookType = TelegramWebhookType
-
-    TELEGRAM_API_BASE = "https://api.telegram.org"
-
-    @classmethod
-    async def validate_payload(
-        cls,
-        webhook: integrations.Webhook,
-        request: Request,
-        credentials: Credentials | None,
-    ) -> tuple[dict, str]:
-        """
-        Validates incoming Telegram webhook request.
-
-        Telegram sends X-Telegram-Bot-Api-Secret-Token header when secret_token
-        was set in setWebhook call.
-
-        Returns:
-            tuple: (payload dict, event_type string)
-        """
-        # Verify secret token header
-        secret_header = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
-        if not secret_header or secret_header != webhook.secret:
-            raise HTTPException(
-                status_code=403,
-                detail="Invalid or missing X-Telegram-Bot-Api-Secret-Token",
-            )
-
-        payload = await request.json()
-
-        # Determine event type based on update content
-        # Telegram updates can contain: message, edited_message,
-        # channel_post, callback_query, etc.
-        if "message" in payload:
-            message = payload["message"]
-            if "text" in message:
-                event_type = "message.text"
-            elif "photo" in message:
-                event_type = "message.photo"
-            elif "voice" in message:
-                event_type = "message.voice"
-            elif "audio" in message:
-                event_type = "message.audio"
-            elif "document" in message:
-                event_type = "message.document"
-            elif "video" in message:
-                event_type = "message.video"
-            else:
-                event_type = "message.other"
-        elif "edited_message" in payload:
-            event_type = "edited_message"
-        elif "message_reaction" in payload:
-            event_type = "message_reaction"
-        elif "callback_query" in payload:
-            event_type = "callback_query"
-        else:
-            event_type = "unknown"
-
-        return payload, event_type
-
-    async def _register_webhook(
-        self,
-        credentials: Credentials,
-        webhook_type: TelegramWebhookType,
-        resource: str,
-        events: list[str],
-        ingress_url: str,
-        secret: str,
-    ) -> tuple[str, dict]:
-        """
-        Register webhook with Telegram using setWebhook API.
-
-        Args:
-            credentials: Bot token credentials
-            webhook_type: Type of webhook (always BOT for Telegram)
-            resource: Resource identifier (unused for Telegram, bots are global)
-            events: Events to subscribe to
-            ingress_url: URL to receive webhook payloads
-            secret: Secret token for request validation
-
-        Returns:
-            tuple: (provider_webhook_id, config dict)
-        """
-        if not isinstance(credentials, APIKeyCredentials):
-            raise ValueError("API key (bot token) is required for Telegram webhooks")
-
-        token = credentials.api_key.get_secret_value()
-        url = f"{self.TELEGRAM_API_BASE}/bot{token}/setWebhook"
-
-        # Telegram setWebhook parameters
-        webhook_data = {
-            "url": ingress_url,
-            "secret_token": secret,
-            "allowed_updates": ["message", "edited_message", "message_reaction"],
-        }
-
-        response = await Requests().post(url, json=webhook_data)
-        result = response.json()
-
-        if not result.get("ok"):
-            error_desc = result.get("description", "Unknown error")
-            raise ValueError(f"Failed to set Telegram webhook: {error_desc}")
-
-        # Telegram doesn't return a webhook ID, use empty string
-        config = {
-            "url": ingress_url,
-            "allowed_updates": webhook_data["allowed_updates"],
-        }
-
-        return "", config
-
-    async def _deregister_webhook(
-        self, webhook: integrations.Webhook, credentials: Credentials
-    ) -> None:
-        """
-        Deregister webhook by calling setWebhook with empty URL.
-
-        This removes the webhook from Telegram's servers.
-        """
-        if not isinstance(credentials, APIKeyCredentials):
-            raise ValueError("API key (bot token) is required for Telegram webhooks")
-
-        token = credentials.api_key.get_secret_value()
-        url = f"{self.TELEGRAM_API_BASE}/bot{token}/setWebhook"
-
-        # Setting empty URL removes the webhook
-        response = await Requests().post(url, json={"url": ""})
-        result = response.json()
-
-        if not result.get("ok"):
-            error_desc = result.get("description", "Unknown error")
-            logger.warning(f"Failed to deregister Telegram webhook: {error_desc}")

autogpt_platform/backend/backend/util/settings.py

@@ -368,6 +368,10 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
         default=600,
         description="The timeout in seconds for Agent Generator service requests (includes retries for rate limits)",
     )
+    agentgenerator_use_dummy: bool = Field(
+        default=False,
+        description="Use dummy agent generator responses for testing (bypasses external service)",
+    )
 
     enable_example_blocks: bool = Field(
         default=False,