Alibaba Cloud DMS MCP Server

Instructions

Implementation Reference

• src/alibabacloud_dms_mcp_server/server.py:273-294 (handler)

  The handler function that implements the core logic of the getDatabase tool by calling the Alibaba Cloud DMS Enterprise API's get_database method with the provided host, port, schema_name, and optional sid.

  async def get_database(
          host: str = Field(description="Hostname or IP of the database instance"),
          port: str = Field(description="Connection port number"),
          schema_name: str = Field(description="Name of the database schema"),
          sid: Optional[str] = Field(default=None, description="Required for Oracle like databases")
  ) -> DatabaseDetail:
      client = create_client()
      req = dms_enterprise_20181101_models.GetDatabaseRequest(host=host, port=port, schema_name=schema_name)
  
      if sid:
          req.sid = sid
      if mcp.state.real_login_uid:
          req.real_login_user_uid = mcp.state.real_login_uid
  
      try:
          resp = client.get_database(req)
          db_data = resp.body.to_map().get('Database', {}) if resp and resp.body else {}
          return DatabaseDetail(**db_data)
      except Exception as e:
          logger.error(f"Error in get_database: {e}")
          raise

• src/alibabacloud_dms_mcp_server/server.py:703-705 (registration)

  Registration of the 'getDatabase' tool in the full toolset using FastMCP's tool decorator, which binds the get_database handler function and provides description and annotations.

  self.mcp.tool(name="getDatabase",
                description="Obtain detailed information about a specific database in DMS when the host and port are provided.",
                annotations={"title": "获取DMS数据库详情", "readOnlyHint": True})(get_database)

• src/alibabacloud_dms_mcp_server/server.py:59-66 (schema)

  Pydantic model defining the structure and field descriptions for the output of the getDatabase tool.

  class DatabaseDetail(MyBaseModel):
      DatabaseId: Any = Field(description="Unique database identifier in DMS", default=None)
      SchemaName: Any = Field(description="Name of the database schema", default=None)
      DbType: Any = Field(description="Database Engine type", default=None)
      InstanceAlias: Any = Field(description="Instance alias in DMS", default=None)
      InstanceId: Any = Field(description="Instance identifier in DMS", default=None)
      State: Any = Field(description="Current operational status", default=None)