feat(connector): wire /types endpoint to catalog with custom_mcp option

- Add ConnectorManager.get_catalog() public accessor
- Replace hardcoded /types response with catalog-driven ConnectorTypeOption list
- Append custom_mcp placeholder for future user-defined MCP servers

packages/dbgpt-core/src/dbgpt/agent/resource/connector/manager.py

@@ -7,8 +7,8 @@ import secrets
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
-from dbgpt.component import BaseComponent, SystemApp
 from dbgpt.agent.resource.tool.base import BaseTool
+from dbgpt.component import BaseComponent, SystemApp
 
 from .catalog import ConnectorCatalog
 from .confirmation import ConfirmationInterceptor, ConfirmationRegistry
@@ -93,6 +93,14 @@ class ConnectorManager(BaseComponent):
     # Catalog helpers
     # ------------------------------------------------------------------
 
+    def get_catalog(self) -> ConnectorCatalog:
+        """Public accessor for the catalog (used by /types endpoint).
+
+        Returns:
+            ConnectorCatalog: The loaded connector catalog.
+        """
+        return self._catalog
+
     def load_catalog(self, path: str) -> None:
         """Load connector type definitions from a *catalog.json* file.
 

packages/dbgpt-core/src/dbgpt/agent/resource/connector/tests/test_manager.py

@@ -4,7 +4,6 @@ import base64
 
 import pytest
 
-from dbgpt.component import SystemApp
 from dbgpt.agent.resource.connector.catalog import (
     AuthConfig,
     AuthField,
@@ -14,6 +13,7 @@ from dbgpt.agent.resource.connector.catalog import (
 from dbgpt.agent.resource.connector.credential import CredentialStore
 from dbgpt.agent.resource.connector.manager import ConnectorManager, ConnectorStatus
 from dbgpt.agent.resource.tool.pack import MCPToolPack
+from dbgpt.component import SystemApp
 
 
 def test_list_active_returns_prompt_ready_connector_summaries():
@@ -83,7 +83,9 @@ def test_list_active_returns_prompt_ready_connector_summaries():
     ]
 
 
-def test_credential_store_uses_stable_encrypt_key_from_env(monkeypatch: pytest.MonkeyPatch):
+def test_credential_store_uses_stable_encrypt_key_from_env(
+    monkeypatch: pytest.MonkeyPatch,
+):
     encrypt_key = base64.urlsafe_b64encode(b"connector-master-key-32-bytes!!").decode()
     monkeypatch.setenv("ENCRYPT_KEY", encrypt_key)
 
@@ -95,7 +97,9 @@ def test_credential_store_uses_stable_encrypt_key_from_env(monkeypatch: pytest.M
     assert store_b.decrypt(encrypted, salt="salt-1") == {"token": "secret-value"}
 
 
-def test_credential_store_prefers_system_app_encrypt_key(monkeypatch: pytest.MonkeyPatch):
+def test_credential_store_prefers_system_app_encrypt_key(
+    monkeypatch: pytest.MonkeyPatch,
+):
     env_key = base64.urlsafe_b64encode(b"env-master-key-32-bytes-value!!!").decode()
     app_key = base64.urlsafe_b64encode(b"app-master-key-32-bytes-value!!!").decode()
     monkeypatch.setenv("ENCRYPT_KEY", env_key)
@@ -106,7 +110,64 @@ def test_credential_store_prefers_system_app_encrypt_key(monkeypatch: pytest.Mon
     store_a = CredentialStore(system_app=system_app)
     encrypted = store_a.encrypt({"token": "secret-value"}, salt="salt-2")
 
-    monkeypatch.setenv("ENCRYPT_KEY", base64.urlsafe_b64encode(b"other-env-key-32-bytes-value!!!").decode())
+    monkeypatch.setenv(
+        "ENCRYPT_KEY",
+        base64.urlsafe_b64encode(b"other-env-key-32-bytes-value!!!").decode(),
+    )
     store_b = CredentialStore(system_app=system_app)
 
     assert store_b.decrypt(encrypted, salt="salt-2") == {"token": "secret-value"}
+
+
+def test_get_catalog_returns_catalog_instance():
+    """get_catalog() should expose the internal ConnectorCatalog."""
+    manager = ConnectorManager()
+
+    catalog = manager.get_catalog()
+
+    assert catalog is manager._catalog
+    assert catalog.list() == []
+
+
+def test_get_catalog_after_load_returns_entries(tmp_path):
+    """get_catalog() should reflect entries loaded via load_catalog()."""
+    import json
+
+    catalog_data = {
+        "connectors": [
+            {
+                "type": "test_conn",
+                "display_name": "Test Connector",
+                "description": "A test connector",
+                "icon": "test",
+                "category": "testing",
+                "mcp_server": {
+                    "server_uri": "http://localhost:9999/sse",
+                    "transport": "sse",
+                },
+                "auth": {
+                    "type": "token",
+                    "fields": [
+                        {
+                            "name": "token",
+                            "label": "Token",
+                            "type": "password",
+                            "required": True,
+                        }
+                    ],
+                    "header_mapping": {"token": "Authorization"},
+                },
+            }
+        ]
+    }
+
+    catalog_path = tmp_path / "catalog.json"
+    catalog_path.write_text(json.dumps(catalog_data))
+
+    manager = ConnectorManager()
+    manager.load_catalog(str(catalog_path))
+    catalog = manager.get_catalog()
+
+    assert len(catalog.list()) == 1
+    assert catalog.list()[0].type == "test_conn"
+    assert catalog.list()[0].display_name == "Test Connector"

packages/dbgpt-serve/src/dbgpt_serve/connector/api/endpoints.py

@@ -1,7 +1,7 @@
 """REST API endpoints for connector management."""
 
 import logging
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel
@@ -16,6 +16,7 @@ from ..service.service import (
     ConnectorService,
     ConnectorUpdateRequest,
 )
+from .schemas import ConnectorTypeOption
 
 logger = logging.getLogger(__name__)
 
@@ -30,9 +31,74 @@ def get_service() -> ConnectorService:
     )
 
 
-@router.get("/types", response_model=Result[List[str]])
-async def list_connector_types() -> Result[List[str]]:
-    return Result.succ(["yuque", "feishu", "notion", "confluence", "github", "gitlab"])
+# Synthetic catalog entry for user-defined custom MCP servers.
+# Kept in sync with CustomMcpForm (Task F) on the frontend.
+# Note: auth_fields here include extra keys (`options`, `default`) beyond the
+# `AuthField` schema in catalog.py — the frontend CustomMcpForm tolerates both shapes.
+_CUSTOM_MCP_OPTION = ConnectorTypeOption(
+    type="custom_mcp",
+    display_name="自定义 MCP Server",
+    description="接入任意 SSE-based MCP Server",
+    category="custom",
+    is_custom=True,
+    auth_fields=[
+        {
+            "name": "server_uri",
+            "label": "SSE Endpoint",
+            "type": "url",
+            "required": True,
+        },
+        {
+            "name": "auth_type",
+            "label": "认证方式",
+            "type": "select",
+            "options": ["none", "bearer", "token"],
+            "required": True,
+        },
+        {
+            "name": "token",
+            "label": "Token",
+            "type": "password",
+            "required": False,
+        },
+        {
+            "name": "header_name",
+            "label": "Token Header 名",
+            "type": "text",
+            "required": False,
+            "default": "Authorization",
+        },
+    ],
+)
+
+
+@router.get("/types", response_model=Result[List[ConnectorTypeOption]])
+async def list_connector_types() -> Result[List[Dict[str, Any]]]:
+    from dbgpt.agent.resource.connector.manager import ConnectorManager as _CM
+
+    if global_system_app is None:
+        raise HTTPException(status_code=503, detail="SystemApp not initialised")
+    manager = global_system_app.get_component(
+        "connector_manager", _CM, default_component=None
+    )
+    if manager is None:
+        raise HTTPException(status_code=503, detail="ConnectorManager not available")
+
+    options: List[ConnectorTypeOption] = []
+    for entry in manager.get_catalog().list():
+        options.append(
+            ConnectorTypeOption(
+                type=entry.type,
+                display_name=entry.display_name,
+                description=entry.description,
+                icon=entry.icon,
+                category=entry.category,
+                is_custom=False,
+                auth_fields=[f.model_dump() for f in entry.auth.fields],
+            )
+        )
+    options.append(_CUSTOM_MCP_OPTION)
+    return Result.succ([o.model_dump() for o in options])
 
 
 @router.post("/", response_model=Result[ConnectorResponse])
@@ -42,7 +108,9 @@ async def create_connector(
 ) -> Result[ConnectorResponse]:
     try:
         return Result.succ(
-            await blocking_func_to_async(global_system_app, service.create_connector, request)
+            await blocking_func_to_async(
+                global_system_app, service.create_connector, request
+            )
         )
     except Exception as e:
         logger.exception("Create connector failed")

packages/dbgpt-serve/src/dbgpt_serve/connector/api/schemas.py

@@ -0,0 +1,21 @@
+"""Pydantic schemas for the connector HTTP API."""
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class ConnectorTypeOption(BaseModel):
+    """Connector type entry returned by GET /types.
+
+    Built-in entries are sourced from the loaded catalog;
+    the `custom_mcp` entry is a synthetic placeholder for user-defined MCP servers.
+    """
+
+    type: str
+    display_name: str
+    description: str
+    icon: Optional[str] = None
+    category: str
+    is_custom: bool = False
+    auth_fields: List[Dict[str, Any]]